Memory
The orbiter-memory package provides typed, protocol-based memory management for agents. It separates memory into short-term (conversation-scoped) and long-term (persistent, extractable) layers, wit...
The orbiter-memory package provides typed, protocol-based memory management for agents. It separates memory into short-term (conversation-scoped) and long-term (persistent, extractable) layers, with a typed hierarchy for different message roles and a status lifecycle for memory curation.
Basic Usage
from orbiter.memory import (
MemoryItem, MemoryMetadata, MemoryStatus,
HumanMemory, AIMemory, SystemMemory, ToolMemory,
ShortTermMemory,
)
# Create typed memories
system_msg = SystemMemory(content="You are a helpful assistant.")
user_msg = HumanMemory(
content="What is Python?",
metadata=MemoryMetadata(user_id="u-1", session_id="s-1"),
)
ai_msg = AIMemory(content="Python is a programming language.", tool_calls=[])
# Short-term memory with session scope
stm = ShortTermMemory(scope="session")
stm.add(system_msg)
stm.add(user_msg)
stm.add(ai_msg)
# Retrieve recent conversation
messages = stm.get_messages(
metadata=MemoryMetadata(session_id="s-1"),
max_rounds=10,
)Memory Type Hierarchy
All memory types extend MemoryItem, which is a Pydantic BaseModel:
| Type | Fields | Use Case |
|---|---|---|
MemoryItem | id, content, memory_type, status, metadata, created_at, updated_at | Base type |
SystemMemory | (base fields) | System instructions |
HumanMemory | (base fields) | User messages |
AIMemory | tool_calls: list[dict] | Assistant responses with optional tool calls |
ToolMemory | tool_call_id, tool_name, is_error | Tool execution results |
# AI message with tool calls
ai_msg = AIMemory(
content="Let me search for that.",
tool_calls=[{
"id": "call_123",
"function": {"name": "search", "arguments": '{"query": "Python"}'},
}],
)
# Tool result
tool_msg = ToolMemory(
content="Python is a high-level programming language...",
tool_call_id="call_123",
tool_name="search",
is_error=False,
)Memory Metadata
MemoryMetadata carries routing information for scoping and filtering:
from orbiter.memory import MemoryMetadata
metadata = MemoryMetadata(
user_id="user-alice",
session_id="session-42",
task_id="task-research",
agent_id="assistant",
)All fields are optional. They are used by the memory store and short-term memory for filtering.
Status Lifecycle
Memory items follow a lifecycle controlled by MemoryStatus:
| Status | Description |
|---|---|
DRAFT | Newly created, pending review |
ACCEPTED | Confirmed and usable |
DISCARD | Marked for removal |
item = HumanMemory(content="Hello")
print(item.status) # MemoryStatus.ACCEPTED (default)
# Transition to a new status
item.transition(MemoryStatus.DISCARD)
print(item.status) # MemoryStatus.DISCARDShort-Term Memory
ShortTermMemory manages recent conversation messages with scoping and windowing:
from orbiter.memory import ShortTermMemory, MemoryMetadata
stm = ShortTermMemory(scope="session")
# Add messages
stm.add(system_msg)
stm.add(user_msg)
stm.add(ai_msg)
# Retrieve with windowing
messages = stm.get_messages(
metadata=MemoryMetadata(session_id="s-1"),
max_rounds=5, # last 5 conversation rounds
)The scope parameter controls filtering:
| Scope | Filters By |
|---|---|
"user" | metadata.user_id |
"session" | metadata.session_id |
"task" | metadata.task_id |
Short-term memory also maintains tool call integrity — when windowing truncates messages, it ensures tool call/result pairs are kept together.
Long-Term Memory
LongTermMemory stores persistent knowledge extracted from conversations:
from orbiter.memory import LongTermMemory, ExtractionType
ltm = LongTermMemory()
# Store extracted knowledge
ltm.add(
content="User prefers Python over JavaScript",
extraction_type=ExtractionType.USER_PROFILE,
metadata=MemoryMetadata(user_id="u-1"),
)Extraction types:
| Type | Description |
|---|---|
USER_PROFILE | User preferences and characteristics |
AGENT_EXPERIENCE | Patterns learned from agent execution |
FACTS | Factual knowledge extracted from conversations |
Long-term memory includes deduplication — adding a memory that is similar to an existing one updates rather than duplicates.
Memory Orchestrator
The MemoryOrchestrator coordinates extraction from conversations into long-term memory:
from orbiter.memory import MemoryOrchestrator, OrchestratorConfig
orchestrator = MemoryOrchestrator(
config=OrchestratorConfig(
batch_size=5,
max_concurrent=3,
),
)
# Submit conversation for extraction
orchestrator.submit(conversation_messages, extractor=my_extractor)
# Process all pending extractions
results = await orchestrator.process_all()Memory Summarization
When conversations grow long, use summarization to compress older messages:
from orbiter.memory import SummaryConfig, SummaryTemplate, check_trigger, generate_summary
config = SummaryConfig(
threshold=20, # trigger after 20 messages
template=SummaryTemplate.CONVERSATION,
)
# Check if summarization should trigger
trigger = check_trigger(messages, config)
if trigger.should_trigger:
summary = await generate_summary(
messages=trigger.messages_to_summarize,
summarizer=my_summarizer,
template=SummaryTemplate.CONVERSATION,
)Summary templates:
| Template | Description |
|---|---|
CONVERSATION | General conversation summary |
FACTS | Extract factual assertions |
PROFILES | Summarize user/agent profiles |
Event Integration
Wrap any MemoryStore with MemoryEventEmitter to emit events on memory operations:
from orbiter.memory import MemoryEventEmitter
from orbiter.events import EventBus
bus = EventBus()
store = SQLiteMemoryStore(":memory:")
emitting_store = MemoryEventEmitter(store=store, bus=bus)
# Events are emitted on add, search, clear
bus.subscribe("memory.added", lambda e: print(f"Added: {e}"))
await emitting_store.add(user_msg)
# prints: Added: ...Events: memory.added, memory.searched, memory.cleared.
Configuration
See Memory Backends for storage backend configuration (SQLite, Postgres, Vector).
API Summary
| Symbol | Module | Description |
|---|---|---|
MemoryItem | orbiter.memory | Base memory type (Pydantic model) |
SystemMemory | orbiter.memory | System instruction memory |
HumanMemory | orbiter.memory | User message memory |
AIMemory | orbiter.memory | Assistant response with tool_calls |
ToolMemory | orbiter.memory | Tool result with tool_call_id, tool_name, is_error |
MemoryMetadata | orbiter.memory | Routing metadata: user_id, session_id, task_id, agent_id |
MemoryStatus | orbiter.memory | Enum: DRAFT, ACCEPTED, DISCARD |
MemoryStore | orbiter.memory | Protocol: add, get, search, clear |
ShortTermMemory | orbiter.memory | Scoped, windowed conversation memory |
LongTermMemory | orbiter.memory | Persistent extracted knowledge with deduplication |
ExtractionType | orbiter.memory | Enum: USER_PROFILE, AGENT_EXPERIENCE, FACTS |
MemoryOrchestrator | orbiter.memory | Coordinates LLM extraction into long-term memory |
SummaryConfig | orbiter.memory | Configuration for summarization triggers |
SummaryTemplate | orbiter.memory | Enum: CONVERSATION, FACTS, PROFILES |
check_trigger | orbiter.memory | Check if summarization should trigger |
generate_summary | orbiter.memory | Generate a summary from messages |
MemoryEventEmitter | orbiter.memory | Event-emitting wrapper for any MemoryStore |