Bulk Ingestion

This walkthrough shows how to load a large dataset into SmartMemory using ingest_batch() for arbitrary items and ingest_conversation() for chat-style data with sessions.

Both APIs run the full 11-stage pipeline per item — they're not shortcuts. They just parallelize the work with asyncio.TaskGroup + Semaphore so you don't pay round-trip latency on every item.

When to use what

API	Input shape	Concurrency cap	Container node
`ingest_batch` / `ingest_batch_sync`	List of strings, dicts, or `MemoryItem`s	`max_concurrent=8`	None
`ingest_conversation` / `ingest_conversation_sync`	List of `{speaker, content}` turns	`max_concurrent=4`	One INDEXED `conversation` node with `PART_OF` edges

Source: smartmemory/batch.py, smartmemory/conversation/bulk_ingest.py.

Bulk-ingesting an arbitrary dataset

"""Bulk-ingest a 1,000-item dataset and verify with search."""
from smartmemory import SmartMemory

memory = SmartMemory()

# 1. Build the items list. Strings, dicts, or MemoryItems all work.
def load_articles(path: str) -> list[str]:
    with open(path) as f:
        # one summary per line for the demo
        return [line.strip() for line in f if line.strip()]

items = load_articles("./articles.txt")  # ~1000 lines
print(f"loaded {len(items)} items")

# 2. Run the batch (sync wrapper — must not be inside an event loop)
response = memory.ingest_batch_sync(items, max_concurrent=8)

# 3. Inspect aggregate metrics
m = response.metrics
print(
    f"ingested {m.succeeded}/{m.total_items} "
    f"in {m.total_duration_ms/1000:.1f}s "
    f"({m.items_per_second:.1f} items/sec, "
    f"{m.rate_limit_hits} rate-limit retries)"
)

# 4. Spot per-item failures
failures = [r for r in response.results if not r.ok]
for f in failures[:5]:
    print(f"FAIL: {f.error}")

# 5. Search to verify it landed
hits = memory.search("renewable energy", top_k=5)
for h in hits:
    print(f"  - {h.content[:80]}")

ingest_batch() doesn't return progress events directly, but each ingest call inside the batch emits to the Progress Event Bus on Redis Streams. Subscribe to GET /memory/progress/stream while the batch runs to watch it live.

For a quick stdout-style progress bar, drive the loop yourself instead of using ingest_batch_sync:

import asyncio
from smartmemory.batch import ingest_batch

async def run_with_progress(memory, items):
    response = await ingest_batch(memory, items, max_concurrent=8)
    return response

async def main():
    response = await run_with_progress(memory, items)
    for i, r in enumerate(response.results):
        status = "ok" if r.ok else f"err: {r.error}"
        print(f"[{i+1}/{len(items)}] {r.duration_ms:.0f}ms  {status}")

asyncio.run(main())

Forwarding kwargs

All ingest() kwargs are forwarded per item:

response = memory.ingest_batch_sync(
    items,
    max_concurrent=8,
    extract_decisions=True,    # auto-extract decisions per item
    extract_reasoning=False,
    context={"source": "research-corpus-2024"},
)

Conversation ingestion

For chat history, use ingest_conversation(). It groups turns into session chunks (each chunk runs the full pipeline once), creates a single conversation container node, and links every chunk back via PART_OF.

"""Import an exported chat into SmartMemory as session-chunked memories."""
from smartmemory import SmartMemory

memory = SmartMemory()

turns = [
    {"speaker": "user", "content": "I'm refactoring the billing service."},
    {"speaker": "assistant", "content": "What's the current architecture?"},
    {"speaker": "user", "content": "Single Postgres, monolith. We want to split."},
    # ... 100s more ...
]

# Optional — explicit session boundaries (turn indices where new sessions start)
session_boundaries = [0, 30, 75]  # 3 sessions

response = memory.ingest_conversation_sync(
    turns,
    session_boundaries=session_boundaries,
    conversation_id="acme-billing-refactor",
    turns_per_chunk=15,        # auto-chunk size when no boundaries given
    max_chunk_chars=12000,     # safety cap per chunk
    max_concurrent=4,
)

print(f"conversation node: {response.conversation_node_id}")
print(f"chunks: {response.chunks_ingested} ok / {response.chunks_failed} failed")
print(f"total turns: {response.total_turns}")
print(f"elapsed: {response.total_duration_ms/1000:.1f}s")

# Each chunk is a regular memory item — search across them as usual
for hit in memory.search("billing service architecture", top_k=3):
    print(f"  - {hit.content[:100]}")

max_concurrent is a hard cap on simultaneous LLM calls. Default 8 for ingest_batch, 4 for ingest_conversation.
Rate-limit retries — the batch runner retries up to 3 times on 429 responses with exponential backoff; surfaced as metrics.rate_limit_hits.
Memory shared across threads — each item runs ingest() in asyncio.to_thread(). The shared SmartMemory instance is safe; cache fills and FalkorDB writes have known benign races (see docs/features/RLM-1e/design.md).

Verifying ingestion

After a large import, sanity-check it:

# Total count via search (sample query)
n_hits = len(memory.search("the", top_k=1000))
print(f"sampled {n_hits} items via broad search")

# Last batch metrics persist on the memory instance
print(f"last_trajectory: {memory.last_trajectory.profile_name if memory.last_trajectory else 'none'}")

For exact counts or graph stats, query the API directly (GET /memory/stats) or the graph backend.

Progress Events — live ingestion feed
Lite Mode — useful for benchmarking bulk ingest locally
Source: smartmemory/batch.py, smartmemory/conversation/bulk_ingest.py

Bulk Ingestion

When to use what

Bulk-ingesting an arbitrary dataset

Watching progress

Forwarding kwargs

Conversation ingestion

Auto-chunking

Session dates

Tuning concurrency

Verifying ingestion

On this page