How Ingestion Works

import uuid
from maximem_synap import MaximemSynapSDK

sdk = MaximemSynapSDK(api_key="your_api_key")

result = await sdk.memories.create(
    document="User: Can you remind me what we decided about the migration timeline?\n"
             "Assistant: In our last conversation, you and the team agreed to begin the "
             "database migration on March 15th, with a two-week buffer for testing.",
    document_type="ai-chat-conversation",
    user_id="user_123",
    customer_id="acme_corp",  # B2B only, auto-resolved on B2C
    mode="fast",
)

print(f"Ingestion ID: {result.ingestion_id}")
# Returns immediately, processing happens in the background

import uuid
from maximem_synap import MaximemSynapSDK
from openai import AsyncOpenAI

sdk = MaximemSynapSDK(api_key="synap_api_key")
openai_client = AsyncOpenAI(api_key="openai_api_key")

# conversation_id must be a valid UUID
conversation_id = str(uuid.uuid4())

async def handle_message(user_message: str, user_id: str, customer_id: str, conversation_id: str):
    """Handle a single user message with memory-enabled context."""

    # 1. Retrieve relevant context from Synap
    context = await sdk.conversation.context.fetch(
        conversation_id=conversation_id,
        user_id=user_id,
        customer_id=customer_id,  # B2B only, auto-resolved on B2C
        search_query=[user_message],
        mode="fast",
    )

    # 2. Build the prompt with retrieved memories
    system_prompt = (
        "You are a helpful assistant. Use the following context from previous "
        "conversations to inform your response:\n\n"
        f"{context.formatted_context}\n\n"
        "If the context is not relevant, respond based on your general knowledge."
    )

    # 3. Generate the response
    response = await openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_message},
        ],
    )
    assistant_message = response.choices[0].message.content

    # 4. Ingest the conversation turn (non-blocking)
    await sdk.memories.create(
        document=f"User: {user_message}\nAssistant: {assistant_message}",
        document_type="ai-chat-conversation",
        user_id=user_id,
        customer_id=customer_id,
        mode="fast",
    )

    return assistant_message

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, auto-resolved 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}")

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, auto-resolved 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