Skip to main content

Memory & Knowledge Graph

Query the character's persistent memory system to retrieve facts, relationships, and conversation history across sessions.

Overview​

Estuary characters remember things about users across conversations. The memory system stores facts, relationships, and context as a knowledge graph. The SDK provides a MemoryClient accessible via client.memory that lets you query this data through the REST API.

Memory is scoped to a specific character-player pair -- each player has their own memory with each character.

For more details on how memory works, see the Memory System platform documentation.

Authentication​

The MemoryClient authenticates using the same API key configured on the EstuaryClient. It becomes available after connect():

async with EstuaryClient(config) as client:
    await client.connect()

    # client.memory is now ready to use
    facts = await client.memory.get_core_facts()
info

The memory API uses REST endpoints, not the WebSocket connection. However, client.memory requires connect() first because the character and player IDs are confirmed during the session handshake.

Listing Memories​

Retrieve a paginated list of memories with optional filtering:

result = await client.memory.get_memories(
    status="active",
    limit=20,
    offset=0,
)

print(f"Total memories: {result.total}")
for memory in result.memories:
    print(f"[{memory.memory_type}] {memory.content}")

Filter Options​

OptionTypeDescription
memory_typestrFilter by memory type
statusstrFilter by status (e.g., "active")
limitintMax results to return (default: 50)
offsetintPagination offset
sort_bystrDeprecated. Accepted for backward compatibility, but no longer sent and has no effect.
sort_orderstrDeprecated. Accepted for backward compatibility, but no longer sent and has no effect.

:::note Sorting arguments are ignored sort_by and sort_order remain in the signature so existing code keeps working, but the SDK no longer puts them on the wire: the memories list endpoint never applied them. Pass them if you like, the result is the same list either way. :::

Core Facts​

Core facts are high-confidence, stable pieces of information the character has learned about the player -- things like their name, preferences, or important details:

result = await client.memory.get_core_facts()

for fact in result.core_facts:
    print(f"{fact.fact_key}: {fact.fact_value}")

Knowledge Graph​

The knowledge graph represents memories as a network of entities and relationships:

graph = await client.memory.get_graph(
    include_entities=True,
    include_character_memories=True,
)

print("Nodes:", len(graph.nodes))
print("Edges:", len(graph.edges))
print("Stats:", graph.stats)

Graph Options​

OptionTypeDefaultDescription
include_entitiesboolNoneInclude entity nodes in the graph
include_character_memoriesboolNoneInclude the character's own memories

Searching Memories​

Search across all memories using natural language:

results = await client.memory.search("likes pizza", limit=5)

print(f"Found {results.total} results for \"{results.query}\"")
for result in results.results:
    print(f"{result.similarity:.2f}: {result.content} (id: {result.memory_id})")

Search uses semantic similarity, so the query does not need to match the stored text exactly. limit accepts 1 to 50 and defaults to 10 when omitted.

Each hit carries three fields: memory_id, similarity, and content. The hit does not include the full memory record, so use get_memories() if you need the rest of its fields.

:::info Search requires a current SDK version Memory search works from this release onward. Earlier versions of the SDK sent the query in a form the API does not accept, so search() could never return results. Upgrade with pip install --upgrade estuary-sdk if your calls are failing. :::

See Search Memories in the REST API reference for the underlying request and response shape.

Timeline​

View memories organized chronologically:

timeline = await client.memory.get_timeline(group_by="week")

print(f"Total memories: {timeline.total_memories}")
for group in timeline.timeline:
    print(f"{group.date}: {len(group.memories)} memories")

Timeline Options​

OptionTypeDescription
start_datestrISO date string for range start
end_datestrISO date string for range end
group_bystr"day", "week", or "month"

Statistics​

Get aggregate statistics about the character's memory for this player:

stats = await client.memory.get_stats()
print(f"Active: {stats.total_active}, Core facts: {stats.core_facts}")
print(f"By type: {stats.by_type}")

Deleting Memories​

Delete all memories for this player-character pair:

result = await client.memory.delete_all(confirm=True)
print(f"Deleted {result.deleted_count} memories")
warning

This is irreversible. The confirm=True parameter is a safety guard.

Example: Memory Dashboard​

import asyncio
from estuary_sdk import EstuaryClient, EstuaryConfig

config = EstuaryConfig(
    server_url="https://api.estuary-ai.com",
    api_key="est_your_api_key",
    character_id="your-character-uuid",
    player_id="user-123",
)

async def main():
    async with EstuaryClient(config) as client:
        await client.connect()

        # Fetch everything concurrently
        facts, graph, stats, timeline = await asyncio.gather(
            client.memory.get_core_facts(),
            client.memory.get_graph(include_entities=True),
            client.memory.get_stats(),
            client.memory.get_timeline(group_by="week"),
        )

        print("=== Core Facts ===")
        for fact in facts.core_facts:
            print(f"  {fact.fact_key}: {fact.fact_value}")

        print("\n=== Knowledge Graph ===")
        print(f"  {len(graph.nodes)} nodes, {len(graph.edges)} edges")

        print("\n=== Stats ===")
        print(f"  Active: {stats.total_active}, Core facts: {stats.core_facts}")

        print("\n=== Timeline ===")
        for group in timeline.timeline:
            print(f"  {group.date}: {len(group.memories)} memories")

asyncio.run(main())

Real-Time Memory Events​

If you want to be notified when new memories are extracted during a conversation, enable realtime_memory in your config:

config = EstuaryConfig(
    server_url="https://api.estuary-ai.com",
    api_key="est_your_api_key",
    character_id="your-character-uuid",
    player_id="user-123",
    realtime_memory=True,
)

async with EstuaryClient(config) as client:
    async def on_memory_updated(event):
        print(f"Extracted {event.memories_extracted} memories, {event.facts_extracted} facts")
        for memory in event.new_memories:
            print(f"  [{memory.memory_type}] {memory.content} (confidence: {memory.confidence})")

    client.on("memory_updated", on_memory_updated)
    await client.connect()

The memory_updated event fires after each bot response when the server finishes extracting memories from the conversation turn.

tip

Real-time memory events are useful for building UIs that show the user what the character has learned, or for triggering application logic based on extracted facts.

Next Steps​

Last updated