> Full SurrealDB documentation index: https://surrealdb.com/docs/llms.txt

# Bulk import

Importing large quantities of documents or knowledge nodes into SurrealDB Agent Memory.

Populating a new Context, or migrating from another system, means ingesting many documents or structured records at once. This page covers uploading documents concurrently, writing structured records as triples on `/facts`, and keeping a large import safe to re-run.

## Uploading many documents

`POST /api/v1/{context_id}/documents` accepts one document per request. For bulk uploads, issue multiple requests concurrently and track their status independently.

### Concurrent upload pattern

```python
import asyncio
from surrealdb.memory import Memory

memory = Memory(context="acme-prod",
    api_key=os.environ["AGENT_MEMORY_API_KEY"])

files = [
    ("returns-policy.pdf", "Returns Policy"),
    ("shipping-guide.pdf", "Shipping Guide"),
    ("warranty-terms.pdf", "Warranty Terms"),
    ("product-manual.pdf", "Product Manual"),
]

async def upload_file(path, title):
    return await memory.documents.upload(
        path,
        title=title,
        scopes=["org/acme"],
    )

docs = await asyncio.gather(*[upload_file(path, title) for path,
    title in files])
doc_ids = [doc.id for doc in docs]
print(f"Queued {len(doc_ids)} documents")
```

```javascript
import { AgentMemory } from "@surrealdb/memory";

const memory = new AgentMemory({ context: "acme-prod",
    apiKey: process.env.AGENT_MEMORY_API_KEY });

const files = [
    { path: "returns-policy.pdf", title: "Returns Policy" },
    { path: "shipping-guide.pdf", title: "Shipping Guide" },
];

const uploads = files.map(({ path, title }) =>
        memory.documents.upload(path, { title, scopes: ["org/acme"] })
);

const docs = await Promise.all(uploads);
const docIds = docs.map(d => d.id);
```

### Polling for completion

After queuing a batch, poll all document IDs until every document is `ready` or `failed`:

```python
async def wait_for_all(doc_ids):
    pending = set(doc_ids)
    failed = []

    while pending:
        await asyncio.sleep(5)
        for doc_id in list(pending):
            doc = await memory.documents.get(doc_id)
            if doc.status == "ready":
                pending.discard(doc_id)
                print(f"{doc_id}: ready")
            elif doc.status == "failed":
                pending.discard(doc_id)
                failed.append((doc_id, doc.error))
                print(f"{doc_id}: failed - {doc.error}")

    return failed

failed = await wait_for_all(doc_ids)
if failed:
    print(f"{len(failed)} documents failed processing")
```

### Rate limiting

SurrealDB Agent Memory applies rate limits per API key. If you are ingesting thousands of documents, introduce a semaphore to limit concurrency:

```python
semaphore = asyncio.Semaphore(10)  # max 10 concurrent uploads

async def upload_with_limit(path, title):
    async with semaphore:
        return await upload_file(path, title)

docs = await asyncio.gather(*[upload_with_limit(path, title) for path,
    title in files])
```

The recommended concurrency ceiling for standard deployments is 10-20 concurrent uploads. Self-hosted deployments can be tuned according to your infrastructure capacity.

## Bulk structured facts

For structured catalogue or policy data you already trust, use **`POST /api/v1/{context_id}/facts`** with `infer: "triples"` (or batch multiple utterances via **`/facts/batch`**). The reconciler persists entities, attributes, and relations in the unified graph with `source.kind = "document"` or operator-provided provenance.

### Triples or documents

Data that already has a structure, such as a product catalogue, a policy table, or a list of settings with the versions that had them, can reach Agent Memory in two ways:

- **As documents.** A language model reads the text and extracts entities, attributes and relations from it.
- **As triples.** A triple is one fact in three parts: an `entity`, a `key`, and either a `value` or a `target` entity, after the subject, predicate and object of RDF. Agent Memory stores each triple as given, with no language model involved.

A triple reads as a short sentence. The subject is the `entity` the fact is about, the predicate is the `key` naming the relationship, and the object is what the subject relates to. The object is a `value` when it is plain data, such as a number or a name, and a `target` when it is another entity that has facts of its own. A time is not one of the three parts: when a relationship held is recorded beside the triple, in `valid_from` and `valid_until`.

| Fact | `entity` (subject) | `key` (predicate) | `value` or `target` (object) | Kind |
| --- | --- | --- | --- | --- |
| Alice is 34 | Alice | `age` | `value`: 34 | Attribute |
| Alice lives in Paris | Alice | `lives_in` | `target`: Paris | Relation |
| The guard stands at the front door from 10 am to 4 pm | Guard | `stands_at` | `target`: front door | Relation, with `valid_from` 10 am and `valid_until` 4 pm |

The third row shows why a time is kept out of the triple: the guard can stand at the front door during the day and sleep in a bed at night, and each of those is a separate triple with its own validity range.

When the records exist before any text does, writing them out as sentences so that document extraction can rebuild them repeats work that is already done. Each document then costs a model call, takes longer to process, and can be read wrongly, for structure that the source data held exactly. Send records like these as triples when the questions are about their current state, and keep document upload for material whose structure is not known in advance, such as reports, notes and transcripts. Where both exist, for example a settings table together with the release notes that explain each change, send the table as triples and the release notes as documents.

### Validity ranges

A relation triple can carry `valid_from` and `valid_until`, the period during which the relation held in the world. This records a fact that started and ended as a single triple. The example below records that SurrealDB 1.1.0 (released 2024-01-09) added the environment variable `SURREAL_WEBSOCKET_MAX_CONCURRENT_REQUESTS` and 2.2.0 (released 2025-02-11) removed it:

```http
POST /api/v1/{context_id}/facts
Content-Type: application/json
Authorization: Bearer <key>

{
  "infer": "triples",
  "scopes": [["version-history"]],
  "triples": [
    { "entity": { "type": "product", "name": "SurrealDB" },
      "key": "has_environment_variable",
      "target": { "type": "concept", "name": "SURREAL_WEBSOCKET_MAX_CONCURRENT_REQUESTS" },
      "valid_from": "2024-01-09T00:00:00Z",
      "valid_until": "2025-02-11T00:00:00Z" },
    { "entity": { "type": "concept", "name": "SURREAL_WEBSOCKET_MAX_CONCURRENT_REQUESTS" },
      "key": "help_text",
      "value": "How many concurrent tasks can be handled on each WebSocket (defaults to 24)" }
  ]
}
```

An entity's `type` must be one of the [fixed entity types](/docs/agent-memory/tuning/ontology-grounding.md#entity-types-are-a-fixed-set), such as `product` or `concept`. A triple with any other type is stored with the type `other`, without an error. Leave out `valid_until` for a relation that still holds. Both fields apply to relations only: an attribute such as `help_text` takes the time it was learned from the request's `observed_at`, and a later triple with the same entity and key supersedes it.

> [!IMPORTANT]
> Recall and chat answer from what holds now. A relation whose `valid_until` has passed is left out of entity listings, `/query` results and the context `/chat` answers from, and the validity dates of a current relation are not shown to the model that writes a `/chat` answer. An entity whose only triple is an ended relation is therefore not found at all. `asOf` does not bring these back, because it selects by when Agent Memory learned a fact rather than by when the fact was true. To answer questions about the past, such as what a release had or when a setting was removed, write the history as documents that each state the full position at their date, and give each document the date it describes as its `observedAt`.

### Registering the scope

A scope has to exist before a triple can be written to it. Create it once with `POST /api/v1/{context_id}/scopes`, before the first batch:

```http
POST /api/v1/{context_id}/scopes
Content-Type: application/json
Authorization: Bearer <key>

{ "path": "version-history" }
```

A request that names a scope that has not been created is rejected with an `unknown scope node` error.

### Batch size

A request returns only after every triple in it is stored, so the time a request takes grows with the number of triples it holds. Batches of a few hundred triples keep each request well inside common client and proxy timeouts, and a batch that fails can be sent again on its own. Writes reconcile by entity and key, so re-sending a batch does not create duplicates (see [Deduplication](#deduplication)).

```http
POST /api/v1/{context_id}/facts
Content-Type: application/json
Authorization: Bearer <key>

{
  "infer": "triples",
  "scopes": [["org/acme"]],
  "triples": [
    { "entity": { "type": "product", "name": "sku_001" },
      "key": "price", "value": "29.99" },
    { "entity": { "type": "product", "name": "sku_001" },
      "key": "belongs_to",
      "target": { "type": "category", "name": "widgets" } }
  ]
}
```

Response:

```json
{
  "mode": "triples",
  "sessionId": "sess_01hy…",
  "turnId": "turn_01hy…",
  "extraction": {
    "turnId": "turn_01hy…",
    "entities": [ { "id": "…", "name": "sku_001", "entityType": "product",
                    "memoryCategory": "knowledge", "isNew": true } ],
    "attributes": [ { "id": "…", "entityId": "…", "key": "price",
                      "value": "29.99", "memoryCategory": "knowledge" } ],
    "relations": [ { "subject": "product/sku_001", "label": "belongs_to",
                     "object": "category/widgets", "memoryCategory": "knowledge" } ],
    "instructions": [],
    "uncertainties": [],
    "corrections": []
  }
}
```

`infer: "triples"` takes the `triples` array directly and runs no LLM. `text` is optional in this mode.

### Python

```python
# Build nodes and relations from your data source
nodes = [
        {"kind": "product", "slug": record["sku"],
        "title": record["name"], "content": record}
    for record in product_catalogue
]

relations = [
        {"in": ("product", record["sku"]), "out": ("category",
        record["category_slug"]), "label": "belongs_to"}
    for record in product_catalogue
    if record.get("category_slug")
]

# Upsert in batches of 200
BATCH_SIZE = 200
for i in range(0, len(nodes), BATCH_SIZE):
    batch_nodes = nodes[i:i + BATCH_SIZE]
    batch_relations = [r for r in relations if any(
        r["in"][1] == n["slug"] for n in batch_nodes
    )]
    triples = [
        {"entity": {"type": "concept", "name": n["slug"]},
         "key": "title", "value": n["title"]}
        for n in batch_nodes
    ] + [
        {"entity": {"type": "concept", "name": r["in"][1]},
         "key": r["label"],
         "target": {"type": "concept", "name": r["out"][1]}}
        for r in batch_relations
    ]
    result = await memory.remember(infer="triples", triples=triples)
    extraction = result["extraction"]
    print(
        f"Batch {i // BATCH_SIZE + 1}: "
        f"entities={len(extraction['entities'])} "
        f"attributes={len(extraction['attributes'])} "
        f"relations={len(extraction['relations'])}"
    )
```

### JavaScript

```javascript
const BATCH_SIZE = 200;

for (let i = 0; i < nodes.length; i += BATCH_SIZE) {
    const batchNodes = nodes.slice(i, i + BATCH_SIZE);
    const triples = batchNodes.map((n) => ({
        entity: { type: "concept", name: n.slug },
        key: "title",
        value: n.title,
    }));
    const result = await memory.remember(null, { infer: "triples", triples });
    console.log(
        `Batch ${i / BATCH_SIZE + 1}: `
        + `entities=${result.extraction.entities.length} `
        + `relations=${result.extraction.relations.length}`,
    );
}
```

## Deduplication

Document uploads are automatically deduplicated by content hash. If the same file is submitted multiple times during a bulk import - for example, because a script is re-run after a partial failure - each duplicate returns the existing document ID with `deduplicated: true` and no reprocessing occurs.

Triple writes reconcile by entity identity: resubmitting a triple for the same `(type, name)` entity and key supersedes the previous value rather than creating a duplicate record.

These properties make bulk imports safe to re-run. A failed or interrupted import can be restarted from the beginning without creating duplicate records.

## Scope assignment on bulk imports

All documents uploaded in a bulk import share the same scope unless you specify it per-document. For mixed-scope imports - for example, some documents are org-level and others are user-level - structure your upload loop to set scope per file:

```python
async def upload_with_scope(item):
    return await memory.documents.upload(
        item["path"],
        title=item["title"],
        scopes=item["scopes"],
    )

items = [
        {"path": "handbook.pdf", "title": "Employee Handbook",
        "scopes": ["org/acme"]},
        {"path": "preferences.json", "title": "User Prefs",
        "scopes": ["org/acme/user/alice"]},
]

docs = await asyncio.gather(*[upload_with_scope(item) for item in items])
```

## Monitoring a large import

For imports of tens of thousands of documents, track overall progress by listing documents with a status filter:

```python
async def import_progress():
    ready = await memory.documents.list(status="ready")
    queued = await memory.documents.list(status="queued")
    failed = await memory.documents.list(status="failed")

    print(f"Ready: {len(ready)}  Queued: {len(queued)}  Failed: {len(failed)}")
```

`GET /documents` filters on `status` and `mimeType` and pages with `page` / `pageSize`. It has no scope filter - the caller's read region already bounds what comes back.

Failed documents should be inspected individually to determine whether the failure is transient (pipeline overload) or permanent (corrupt file, unsupported format):

```python
failed_docs = await memory.documents.list(status="failed")
for doc in failed_docs:
    full = await memory.documents.get(doc.id)
    print(f"{doc.id}  {full.title}  {full.error}")
```