Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.maximem.ai/llms.txt

Use this file to discover all available pages before exploring further.

When you turn Synap on for an existing app, you usually have months of conversation logs sitting in a database. Pump them in via batch ingestion so the agent starts day one with the same context your users already remember about themselves. 
import asyncio
from typing import AsyncIterator
from maximem_synap import MaximemSynapSDK
from maximem_synap.memories.models import CreateMemoryRequest

sdk = MaximemSynapSDK()
await sdk.initialize()

@dataclass
class HistoricalTurn:
    user_id: str
    customer_id: str
    conversation_id: str
    user_message: str
    assistant_message: str
    happened_at: datetime    # original timestamp from your DB


async def load_historical_turns() -> AsyncIterator[HistoricalTurn]:
    """Pull from your application database. Yield one turn at a time."""
    async for row in db.stream("SELECT * FROM conversations ORDER BY created_at"):
        yield HistoricalTurn(
            user_id=row.user_id,
            customer_id=row.customer_id,
            conversation_id=str(row.conversation_id),   # must be UUID
            user_message=row.user_text,
            assistant_message=row.assistant_text,
            happened_at=row.created_at,
        )


async def batch_replay(batch_size: int = 50):
    batch: list[CreateMemoryRequest] = []
    total = 0

    async for turn in load_historical_turns():
        batch.append(CreateMemoryRequest(
            document=f"User: {turn.user_message}\nAssistant: {turn.assistant_message}",
            document_type="ai-chat-conversation",
            document_created_at=turn.happened_at,   # preserve original time
            user_id=turn.user_id,
            customer_id=turn.customer_id,
            mode="long-range",                       # deep extraction for historical data
            metadata={
                "conversation_id": turn.conversation_id,
                "source": "backfill",
            },
        ))

        if len(batch) >= batch_size:
            await sdk.memories.batch_create(documents=batch, fail_fast=False)
            total += len(batch)
            print(f"backfilled {total} turns")
            batch.clear()

    if batch:
        await sdk.memories.batch_create(documents=batch, fail_fast=False)
        total += len(batch)

    print(f"done: {total} historical turns ingested")

asyncio.run(batch_replay())
Why mode="long-range" for backfill Historical conversations are the highest-value extraction target you’ll ever have — they’re the one-shot chance to build a rich entity graph and detailed preferences for every user. long-range runs the full extraction pipeline (entity resolution, preference detection, emotion analysis, relationship mapping). It’s slower per document, but speed doesn’t matter for an offline backfill. fast mode would skip most of this — fine for runtime chat ingestion, wrong for backfill. Why document_created_at Synap uses document_created_at as the ground-truth timestamp for temporal reasoning, recency ranking, and aging. Don’t skip it — without it, every memory looks like it happened today, and recency-weighted retrieval breaks. Pass the original timestamp from your database. Why batch and not one-by-one batch_create accepts up to 100 documents per call, and the cloud queues them with BOOTSTRAP priority — separate from real-time ingestion lanes, so a backfill doesn’t slow down live user traffic. One-by-one would 30x your wall-clock time and burn rate limits. Idempotency If you re-run the backfill (because you found a bug, or because you’re bringing online a second region), pass a stable document_id derived from your DB row: 
document_id=f"backfill:{turn.conversation_id}:{turn.message_index}",
Synap deduplicates on document_id and returns a 409 for duplicates. Catch and skip: 
try:
    await sdk.memories.batch_create(documents=batch, fail_fast=False)
except SynapError as e:
    # batch_create with fail_fast=False returns per-document results; check each
    ...
How long it takes Rule of thumb: 100K turns at long-range takes a few hours, gated by per-Instance throughput. Talk to support if you have multi-million-turn backfills and want a higher BOOTSTRAP throughput allocation for a window.