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
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")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:
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:
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, akey, and either avalueor atargetentity, 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:
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, 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.
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:
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).
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:
{
"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
# 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
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:
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:
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):
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}")