LongTermMemory API Reference

API for entity, preference, and fact storage.

Overview

LongTermMemory handles the knowledge graph: entities (POLE+O), user preferences, and facts with relationships.

async with MemoryClient(settings) as client:
    # Access via client
    long_term = client.long_term

    # Add an entity
    entity, dedup_result = await long_term.add_entity(
        "John Smith",
        "PERSON",
        subtype="INDIVIDUAL"
    )

Entity Operations

add_entity

Add an entity to the knowledge graph.

entity, dedup_result = await long_term.add_entity(
    name: str,
    entity_type: str,
    subtype: str | None = None,
    description: str | None = None,
    metadata: dict[str, Any] | None = None,
    confidence: float = 1.0,
    coordinates: tuple[float, float] | None = None,
    geocode: bool = False,
    deduplicate: bool = True,
)

Parameters

Parameter Type Default Description

name

str

required

Entity name

entity_type

str

required

POLE+O type: PERSON, OBJECT, LOCATION, EVENT, ORGANIZATION

subtype

str

None

Subtype (e.g., VEHICLE, COMPANY)

description

str

None

Entity description

metadata

dict

None

Additional metadata

confidence

float

1.0

Confidence score (0.0-1.0)

coordinates

tuple

None

Latitude, longitude for locations

geocode

bool

False

Auto-geocode location entities

deduplicate

bool

True

Check for duplicates

Returns

Tuple of (Entity, DeduplicationResult):

@dataclass
class DeduplicationResult:
    action: str  # "created", "merged", "flagged"
    matched_entity_id: str | None
    matched_entity_name: str | None
    confidence: float | None
    match_type: str | None  # "embedding", "fuzzy", "both"

get_entity

Get an entity by ID.

entity = await long_term.get_entity(entity_id: str)

get_entity_by_name

Get an entity by name.

entity = await long_term.get_entity_by_name(
    name: str,
    entity_type: str | None = None,
)

search_entities

Semantic search for entities.

results = await long_term.search_entities(
    query: str,
    entity_type: str | None = None,
    limit: int = 10,
    min_score: float = 0.0,
)

update_entity

Update an entity’s properties.

entity = await long_term.update_entity(
    entity_id: str,
    description: str | None = None,
    metadata: dict[str, Any] | None = None,
)

delete_entity

Delete an entity.

await long_term.delete_entity(entity_id: str)

Preference Operations

add_preference

Add a user preference.

preference = await long_term.add_preference(
    category: str,
    value: str,
    user_id: str | None = None,
    importance: float = 0.5,
    metadata: dict[str, Any] | None = None,
)

Parameters

Parameter Type Default Description

category

str

required

Preference category (e.g., food, style)

value

str

required

Preference value/description

user_id

str

None

Associated user ID

importance

float

0.5

Importance score (0.0-1.0)

metadata

dict

None

Additional metadata

get_preferences

Get preferences, optionally filtered.

preferences = await long_term.get_preferences(
    user_id: str | None = None,
    category: str | None = None,
    limit: int = 100,
)

delete_preference

Delete a preference.

await long_term.delete_preference(preference_id: str)

Fact Operations

add_fact

Add a fact (statement with optional validity period).

fact = await long_term.add_fact(
    subject: str,
    predicate: str,
    object: str,
    confidence: float = 1.0,
    valid_from: datetime | None = None,
    valid_until: datetime | None = None,
)

get_facts

Get facts about a subject.

facts = await long_term.get_facts(
    subject: str | None = None,
    predicate: str | None = None,
    limit: int = 100,
)

Deduplication Operations

find_potential_duplicates

Find entities flagged as potential duplicates.

duplicates = await long_term.find_potential_duplicates(limit: int = 100)
# Returns: [(entity1, entity2, confidence), ...]

review_duplicate

Confirm or reject a duplicate pair.

await long_term.review_duplicate(
    source_id: str,
    target_id: str,
    confirm: bool,  # True to merge, False to reject
)

get_same_as_cluster

Get all entities in a SAME_AS cluster.

cluster = await long_term.get_same_as_cluster(entity_id: str)

Provenance Operations

link_entity_to_message

Link an entity to its source message.

await long_term.link_entity_to_message(
    entity: Entity,
    message_id: str,
    confidence: float = 1.0,
    start_pos: int | None = None,
    end_pos: int | None = None,
    context: str | None = None,
)

link_entity_to_extractor

Link an entity to the extractor that created it.

await long_term.link_entity_to_extractor(
    entity: Entity,
    extractor_name: str,
    confidence: float = 1.0,
    extraction_time_ms: float | None = None,
)

get_entity_provenance

Get provenance information for an entity.

provenance = await long_term.get_entity_provenance(entity: Entity)
# Returns: {"sources": [...], "extractors": [...]}

Geospatial Operations

search_locations_near

Find locations near a point.

locations = await long_term.search_locations_near(
    latitude: float,
    longitude: float,
    radius_km: float = 10.0,
    limit: int = 10,
)

geocode_locations

Batch geocode location entities.

stats = await long_term.geocode_locations(skip_existing: bool = True)
# Returns: {"processed": 100, "geocoded": 85, "skipped": 10, "failed": 5}

Entity Model

@dataclass
class Entity:
    id: str
    name: str
    entity_type: str
    subtype: str | None
    description: str | None
    metadata: dict[str, Any]
    confidence: float
    created_at: datetime
    updated_at: datetime
    embedding: list[float] | None
    aliases: list[str]

    @property
    def full_type(self) -> str:
        """Returns 'TYPE:SUBTYPE' or just 'TYPE'."""

Example

async with MemoryClient(settings) as client:
    # Add entities
    person, _ = await client.long_term.add_entity(
        "John Smith", "PERSON", subtype="INDIVIDUAL"
    )
    company, _ = await client.long_term.add_entity(
        "Acme Corp", "ORGANIZATION", subtype="COMPANY"
    )

    # Add preference
    await client.long_term.add_preference(
        "food", "Loves Italian cuisine", importance=0.9
    )

    # Search entities
    results = await client.long_term.search_entities("software company")
    for entity, score in results:
        print(f"[{score:.2f}] {entity.name} ({entity.full_type})")

See Also