Import existing history (bootstrap ingestion)

Bootstrap ingestion loads pre-existing data into Synap in bulk. Before your agent goes live (or alongside live operation) you often need to seed it with historical context: past conversations, product documentation, knowledge base articles, customer records. You call sdk.memories.batch_create() with many documents at once. Because this data is historical and processed in the background, long-range is the natural default. Use bootstrap ingestion whenever you need to load a significant volume of existing data:

Migrating from another system: a custom memory solution, a competing product, or an in-house knowledge base.
Loading historical conversations: past chat logs, support tickets, or email threads.
Seeding product documentation: docs, FAQs, help center articles, internal wikis.
Backfilling customer data: CRM records, customer profiles, organizational context.
Populating shared knowledge: company policies, SOPs, reference material at customer or client scope.

Bootstrap loads run at BOOTSTRAP priority in the ingestion queue, which processes below real-time but above maintenance tasks. Your live agent keeps operating normally while historical data is processed. You do not need to finish bootstrapping before going live. The two paths can run simultaneously without competing for resources.

The batch ingestion method

sdk.memories.batch_create() accepts multiple documents in a single call, reducing per-call overhead and enabling server-side throughput optimizations. Each document is a CreateMemoryRequest supporting the same fields as sdk.memories.create().

from maximem_synap import MaximemSynapSDK, CreateMemoryRequest

sdk = MaximemSynapSDK(api_key="your_api_key")

result = await sdk.memories.batch_create(
    documents=[
        CreateMemoryRequest(
            document="User: How do I reset my password?\nAssistant: Go to Settings > Security > Reset Password.",
            document_type="ai-chat-conversation",
            document_id="migration_001",
            document_created_at="2024-03-15T10:30:00Z",
            user_id="user_456",
            customer_id="acme_corp",  # B2B only; not accepted on B2C
            mode="long-range",
        ),
        CreateMemoryRequest(
            document="User: What integrations do you support?\nAssistant: We support Slack, Jira, and GitHub.",
            document_type="ai-chat-conversation",
            document_id="migration_002",
            document_created_at="2024-03-16T14:20:00Z",
            user_id="user_789",
            customer_id="acme_corp",
            mode="long-range",
        ),
    ],
    fail_fast=False,
)

print(f"Succeeded: {result.succeeded}")
print(f"Failed: {result.failed}")

Parameter	Required	Description
`documents`	Yes	List of `CreateMemoryRequest` objects to ingest (max 100 per call).
`fail_fast`	No	If `True`, the whole batch fails on the first error. If `False` (default), errors are collected and returned alongside successful results.

Each CreateMemoryRequest supports the same fields as a single create call, including document_id for idempotency and document_created_at for temporal accuracy.

Key considerations

Preserve original timestamps

Always set document_created_at to the document’s original creation time. Without it, Synap defaults to the ingestion time, which distorts temporal ordering. If a user later asks “What did we discuss last March?”, accurate timestamps are essential for correct retrieval.

Use document IDs for idempotency

Assign a unique document_id to every document, ideally derived from your source system’s primary key (e.g. migration_{source_id}). If a batch is interrupted, you can safely retry it: already-ingested documents are skipped, preventing duplicates and making it easy to trace memories back to their source.

Use long-range mode for historical data

Bootstrap data benefits from thorough extraction. long-range (the default for batch) performs deep entity resolution, relationship mapping, and preference detection. The extra processing time is fine for background loads.

Validate and organize before loading

Clean your historical data first: drop empty conversations, strip PII you should not store, ensure timestamps are ISO 8601, and confirm each document carries the correct scope. Incorrect scoping is difficult to fix later: you would have to re-ingest. Start with a small test batch and verify scoping, timestamps, and entity resolution before the full load.

Monitor ingestion progress

For large loads, track progress with sdk.memories.status():

status = await sdk.memories.status(ingestion_id=result.results[0].ingestion_id)

print(f"Status: {status.status}")              # queued | processing | completed | failed
print(f"Memories created: {status.memories_created}")
print(f"Error: {status.error_message}")

const status = await sdk.memories.status(result.results[0]?.ingestion_id);

console.log(`Status: ${status.status}`);  // queued | processing | completed | failed
console.log(`Memories created: ${status.memories_created}`);
console.log(`Error: ${status.error_message}`);

const status = await sdk.memories.status(result.results[0]?.ingestion_id);

console.log(`Status: ${status.status}`);  // queued | processing | completed | failed
console.log(`Memories created: ${status.memories_created}`);
console.log(`Error: ${status.error_message}`);

Keep concurrency modest. While the API accepts many parallel batch requests, excessive concurrency causes queue backpressure and slows processing for all ingestion types. Add a short delay between batches during the initial bulk load.

Full example: loading historical conversations

This loads historical conversations from a database, handling pagination, error recovery, and progress monitoring:

import asyncio
from maximem_synap import MaximemSynapSDK, CreateMemoryRequest

sdk = MaximemSynapSDK(api_key="your_api_key")

BATCH_SIZE = 100

async def load_historical_conversations(db_connection):
    """Load historical conversations from a database into Synap."""

    conversations = await db_connection.fetch(
        "SELECT id, content, user_id, customer_id, created_at "
        "FROM conversations ORDER BY created_at ASC"
    )

    total = len(conversations)
    processed = failed = 0
    ingestion_ids = []

    for i in range(0, total, BATCH_SIZE):
        batch = conversations[i : i + BATCH_SIZE]

        documents = [
            CreateMemoryRequest(
                document=conv["content"],
                document_type="ai-chat-conversation",
                document_id=f"migration_{conv['id']}",
                document_created_at=conv["created_at"].isoformat(),
                user_id=conv["user_id"],
                customer_id=conv["customer_id"],  # B2B only; not accepted on B2C
                mode="long-range",
            )
            for conv in batch
        ]

        try:
            result = await sdk.memories.batch_create(documents=documents, fail_fast=False)
            processed += result.succeeded
            failed += result.failed
            ingestion_ids.extend(r.ingestion_id for r in result.results)
            print(f"Progress: {processed}/{total} ingested, {failed} failed")

            for r in result.results:
                if r.status == "failed":
                    print(f"  Error: {r.error_message}")

        except Exception as e:
            print(f"Batch request failed: {e}")
            # Safe to retry, document_id ensures idempotency
            failed += len(batch)

        # Brief pause between batches to avoid queue backpressure
        await asyncio.sleep(1)

    print(f"\nBootstrap complete: {processed} ingested, {failed} failed")
    return ingestion_ids

Get started

Build with AI

Switch to Synap

Evaluate & pricing

Import existing history (bootstrap ingestion)

The batch ingestion method

Key considerations

Full example: loading historical conversations

​The batch ingestion method

​Key considerations

​Full example: loading historical conversations

The batch ingestion method

Key considerations

Full example: loading historical conversations