deeplake

Deeplake 托管服务 SDK

用于将数据摄取到 Deeplake 托管表中的 Agent 友好型 SDK。 当用户想要在 Deeplake 中存储、摄取或查询数据时，请使用此技能。 提供 Python 和 Node.js/TypeScript 版本。

快速参考

Python

pip install deeplake # uv add deeplake 

Python 导入（主要方式）：

from deeplake import Client

# Async variant (requires aiohttp: pip install aiohttp):
from deeplake.managed import AsyncClient

from deeplake import Client

# Initialize -- token from DEEPLAKE_API_KEY env var, workspace defaults to "default"
client = Client()
client = Client(token="dl_xxx", workspace_id="my-workspace")

# Ingest files (FILE schema)
client.ingest("videos", {"path": ["video1.mp4", "video2.mp4"]}, schema={"path": "FILE"})

# Ingest structured data with indexes for search
client.ingest("embeddings", {
    "text": ["doc1", "doc2", "doc3"],
    "embedding": [[0.1, 0.2, ...], [0.3, 0.4, ...], [0.5, 0.6, ...]],
}, index=["embedding", "text"])

# Ingest from HuggingFace
client.ingest("cifar", {"_huggingface": "cifar10"})

# Ingest with format object (see formats.md for CocoPanoptic, Coco, LeRobot, custom)
client.ingest("table", format=my_format)

# Fluent query
results = client.table("videos").select("id", "text").where("file_id = $1", "abc").limit(10)()

# Raw SQL
results = client.query("SELECT * FROM videos LIMIT 10")

# Vector similarity search -- through client.query() the vector must be
# INLINED, not passed as $1 (see "Querying" below for why)
from deeplake.managed import vector_literal
results = client.query(f"""
    SELECT id, text, embedding::float4[] <#> {vector_literal(query_embedding)} AS similarity
    FROM embeddings ORDER BY similarity DESC LIMIT 10
""")

# Create an empty table with an explicit schema (no data yet)
client.create_table("docs", {
    "id": "TEXT",
    "embedding": "EMBEDDING(1536)",     # NOT bare "EMBEDDING" — keep the dimension
}, index={"id": "inverted_index", "embedding": "clustered"})

# Table management
client.list_tables()
client.drop_table("old_table")
client.create_index("embeddings", "embedding", index_type="clustered")
client.drop_index("embeddings", "embedding")
client.describe_table("embeddings")     # {col: {"type": ..., "index": ...}}

# Open a table without taking write access
ds = client.open_table("embeddings", read_only=True)

Node.js / TypeScript

npm install deeplake

TypeScript 导入：

import { ManagedClient, initializeWasm } from 'deeplake';

WASM 初始化（任何操作前都必须执行）：

await initializeWasm();

在启动时、执行任何 ManagedClient 操作（摄取、查询等）之前，调用一次 initializeWasm()。它会初始化底层 WASM 模块。

import { ManagedClient, initializeWasm } from 'deeplake';

await initializeWasm();

const client = new ManagedClient({ token: 'dl_xxx', workspaceId: 'my-workspace' });

// Ingest files (FILE schema)
await client.ingest("videos", { path: ["video1.mp4"] }, { schema: { path: "FILE" } });

// Ingest structured data
await client.ingest("embeddings", {
    text: ["doc1", "doc2"],
    embedding: [[0.1, 0.2], [0.3, 0.4]],
});

// Ingest with format object (see formats.md)
await client.ingest("table", null, { format: myFormat });

// Fluent query (use .execute())
const results = await client.table("videos")
    .select("id", "text").where("file_id = $1", "abc").limit(10).execute();

// Raw SQL
const rows = await client.query("SELECT * FROM videos LIMIT 10");

// Create an empty table with an explicit schema (no data yet)
await client.createTable("docs", {
    id: "TEXT",
    embedding: "EMBEDDING(1536)",       // NOT bare "EMBEDDING" — keep the dimension
}, { index: { id: "inverted_index", embedding: "clustered" } });

// Table management
await client.listTables();
await client.dropTable("old_table");
await client.createIndex("embeddings", "embedding", { indexType: "clustered" });
await client.dropIndex("embeddings", "embedding");
await client.describeTable("embeddings");   // {col: {type, index}}

// Open a table without taking write access
const ds = await client.openTable("embeddings", true);

---

依赖项和先决条件

必需服务：

• Deeplake API 服务正在运行（默认值：https://api.deeplake.ai）

可选 Python 依赖项（按文件类型）：

• 视频摄取：ffmpeg（sudo apt-get install ffmpeg）
• PDF 摄取：pymupdf（pip install pymupdf）
• 缩略图生成：Pillow（pip install Pillow）
• COCO 检测格式：pycocotools、Pillow、numpy（pip install pycocotools Pillow numpy）
• LeRobot 帧格式：pandas、numpy（pip install pandas numpy）

可选 TypeScript 依赖项（按文件类型）：

• 视频摄取：ffmpeg（系统二进制文件）
• PDF 摄取：pdfjs-dist（npm install pdfjs-dist）
• 缩略图生成：sharp（npm install sharp）
• COCO 检测格式：无需外部依赖（纯 JS 遮罩渲染）

架构

Python:  Client(token, workspace_id)
Node.js: ManagedClient({ token, workspaceId })
  |-- .ingest(table, data)       -> creates PG table via API, opens al://{ws}/{table}
  |                                 via deeplake SDK (auto credential rotation)
  |-- .create_table(table, schema) -> POST /workspaces/{id}/tables (empty table, no rows)
  |-- .query(sql)                -> POST /workspaces/{id}/tables/query -> list[dict] / QueryRow[]
  |-- .table(table)...           -> fluent SQL builder -> list[dict] / QueryRow[]
  |-- .create_index(table, col, index_type=) -> CREATE INDEX USING deeplake_index
  |-- .drop_index(table, col)    -> DROP INDEX (needed to change a column's index type)
  |-- .describe_table(table)     -> {col: {type, index}} read back from storage
  |-- .open_table(table, read_only=) -> deeplake.open[_read_only]("al://{ws}/{table}")
  |-- .list_tables()             -> GET /workspaces/{id}/tables -> list[str] / string[]
  `-- .drop_table(table)         -> DELETE /workspaces/{id}/tables/{name}
                    |
                    v
              REST API -> PostgreSQL + pg_deeplake
  - All DB operations go through the REST API (no direct PG connection)
  - Dataset access uses al:// paths with automatic credential resolution
  - Creds endpoint: GET /api/org/{workspace}/ds/{table}/creds
  - Vector similarity: embedding::float4[] <#> query_vec
  - BM25 text search:  text <#> 'search query'
  - Hybrid search:     (embedding, text)::deeplake_hybrid_record

---

客户端初始化

Python

from deeplake import Client

client = Client(
    token: str = None,           # API token (falls back to DEEPLAKE_API_KEY env var)
    workspace_id: str = "default",  # Target workspace (default: "default")
    api_url: str = None,         # API URL (default: https://api.deeplake.ai)
    org_id: str = None,          # Target org UUID. REQUIRED when the token's embedded
                                 #   org differs from the org you want to act in (see below).
)

跨组织调用——很容易漏掉 org_id=。 API 令牌中嵌入了一个单一 org_id 声明，即使用户属于多个组织也是如此。SDK 发出的每个请求都限定在某个组织范围内；如果省略 org_id，该范围会静默地默认为令牌中嵌入的组织，而不一定是你想要的组织。只要令牌所属组织 ≠ 目标组织，就传入 org_id=（Python/JS 关键字参数），或设置 ACTIVELOOP_ORG_ID；SDK 会将其转发为 X-Activeloop-Org-Id HTTP 标头。服务器只认可这个确切的标头——其他类似组织的标头名称（例如 X-Org-Id）会被忽略，因此值错误或缺失会导致资源创建在令牌所属组织中，且不会报错。通过 GET /me 将组织名称 → UUID（该接口会返回令牌可访问的组织 ID）。完整的身份验证提供方 / org_id 约定请参阅 reference.md。

Node.js / TypeScript

import { ManagedClient, initializeWasm } from 'deeplake';

await initializeWasm();

const client = new ManagedClient({
    token: string,               // API token (required)
    workspaceId?: string,        // Target workspace (default: "default")
    apiUrl?: string,             // API URL (default: https://api.deeplake.ai)
});

**令牌：**通过 Deeplake 平台创建 API 令牌，地址为 https://app.deeplake.ai/<org_name>/workspace/<workspace>/apitoken。该令牌是嵌入了 org_id 的 JWT。若未设置，则回退到 DEEPLAKE_API_KEY 环境变量（仅限 Python）。

**后端端点：**客户端会在每次打开数据集之前（而非初始化时）将 C++ 后端端点设置为 api_url，使 al:// 路径解析（凭据获取）通过 deeplake-api，而不是旧版 controlplane。这样可避免多个客户端使用不同 API URL 时发生全局状态覆盖。Python：deeplake.client.endpoint = api_url。Node.js：deeplakeSetEndpoint(apiUrl)。

连接生命周期：

# Python: just create and use -- no connection to manage
client = Client()
client.ingest("table", {"path": ["file.txt"]}, schema={"path": "FILE"})
# No close() method -- client is stateless (REST API calls only)

---

数据摄取

Python：client.ingest()

result = client.ingest(
    table_name: str,                    # Table name to create (must not already exist)
    data: dict[str, list] = None,       # Data dict (required unless format= is set).
                                        #   {"_huggingface": "name"} -> HuggingFace dataset
                                        #   schema has "FILE" cols -> file paths processed
                                        #   otherwise -> column data {col: [values]}
    *,
    format: Format = None,              # Format object (subclass of Format) with
                                        #   normalize() method. When set, data is ignored.
                                        #   e.g. CocoPanoptic(images_dir=..., ...)
    schema: dict[str, str] = None,      # Schema override {col: type}
                                        #   Use "FILE" for columns containing file paths
                                        #   See reference.md for all type names
    index: list[str] = None,            # Columns to create deeplake_index on after ingestion.
                                        #   Use for EMBEDDING (vector search) and TEXT (BM25) columns.
    on_progress: Callable = None,       # Progress callback(rows_written, total)
    chunk_size: int = 1000,             # Text chunk size (chars)
    chunk_overlap: int = 200,           # Text chunk overlap (chars)
    pdf_dpi: int = 150,                 # PDF render DPI (higher = sharper but slower)
) -> dict

Node.js：client.ingest()

const result = await client.ingest(
    tableName: string,                              // Table name
    data?: Record<string, unknown[]> | null,        // Data dict (or null when using format)
    options?: {
        format?: Format,                            // Format object with normalize()
        schema?: Record<string, string>,            // Schema override
        index?: string[],                           // Columns to create deeplake_index on
        onProgress?: (processed, total) => void,    // Progress callback
        chunkSize?: number,                         // Text chunk size (default 1000)
        chunkOverlap?: number,                      // Text chunk overlap (default 200)
    },
): Promise<IngestResult>

表是否存在： 如果 table_name 已存在，ingest() 会将数据追加到现有表中，不会删除并重建该表。若要替换现有表，请先调用 client.drop_table(table_name)。PG 表的架构必须与要追加的新数据兼容。

返回： {"table_name": "videos", "row_count": 150, "dataset_path": "al://workspace_id/videos"}

⚠️ 务必将 row_count 与发送的数据量进行核对。 ingest() 返回时没有抛出错误，并不能证明数据行已提交。在后端凭据刷新或资源配置期间，批次可能在 HTTP 层被接受，但最终写入 0 行；旧版服务即使写入成功，也可能返回 row_count: 0。请以 row_count 为准：如果它不等于 len(data)，请重试缺失的数据行（ingest 是追加操作，因此请使用确定性的主键并去重，或使用 INSERT … ON CONFLICT DO NOTHING，以确保操作幂等）。大量写入后，请使用 SELECT COUNT(*) FROM <table> 确认结果，但请注意可见性延迟（参见查询 → 读己之写）。

同时提供 data 和 format： 如果两者都已提供，则以 format 为准，并忽略 data。如果两者都未提供，则会抛出 IngestError。

缩略图： 当格式对象声明了 image_columns()（pg_schema 类型为 "IMAGE" 的列）时，系统会自动生成 4 种尺寸（32x32、64x64、128x128、256x256）的缩略图，并将其存储在共享的 thumbnails 数据集中，路径为 {root_path}/thumbnails。需要 Pillow（Python）或 sharp（Node.js）。

按文件类型划分的分块策略

文件类型	扩展名	策略	创建的列
视频	.mp4、.mov、.avi、.mkv、.webm	10 秒片段 + 缩略图	id、file_id、chunk_index、start_time、end_time、video_data、thumbnail、text
图像	.jpg、.jpeg、.png、.gif、.bmp、.webp	单个分块	id、file_id、image、filename、text
PDF	.pdf	按页处理，分辨率为 150 DPI（可通过 pdf_dpi 配置）	id、file_id、page_index、image、text
文本	.txt、.md、.csv、.json、.xml、.html	每块 1000 个字符，重叠 200 个字符	id、file_id、chunk_index、text
其他	*	单个二进制分块	id、file_id、data、filename

示例

# Ingest files (FILE schema)
client.ingest("videos", {"path": ["cam1.mp4", "cam2.mp4"]}, schema={"path": "FILE"})
client.ingest("photos", {"path": ["img1.jpg"]}, schema={"path": "FILE"})
client.ingest("manuals", {"path": ["manual.pdf"]}, schema={"path": "FILE"})

# Ingest structured data (dict = column data, schema inferred)
client.ingest("vectors", {
    "text": ["Hello", "Goodbye"],
    "embedding": [[0.1, 0.2, 0.3], [0.4, 0.5, 0.6]],
})

# Ingest with explicit schema
client.ingest("data", {"name": ["Alice", "Bob"], "age": [30, 25]},
              schema={"name": "TEXT", "age": "INT64"})

# JSONB is a supported schema type — store arbitrary nested JSON per row.
# (Not obvious from the type table; confirmed working end-to-end.)
client.ingest("events", {"id": [1, 2], "payload": [{"k": "v"}, {"n": [1, 2]}]},
              schema={"id": "INT64", "payload": "JSONB"})

# Ingest from HuggingFace
client.ingest("mnist", {"_huggingface": "mnist"})

# Ingest with a format object (see formats.md for CocoPanoptic, Coco, LeRobot, custom)
client.ingest("table", format=my_format)

# Ingest with progress callback
def progress(rows_written, total):
    print(f"Written {rows_written} rows...")
client.ingest("docs", {"path": pdf_files}, schema={"path": "FILE"}, on_progress=progress)

自定义格式类请参见 formats.md。 更多摄取示例请参见 examples.md。

---

训练 / 流式处理

client.open_table()

将托管表作为 deeplake.Dataset 打开以直接访问——绕过 PostgreSQL，并返回原生数据集对象，该对象内置了 ML 框架集成。

ds = client.open_table(table_name: str, read_only: bool = False) -> deeplake.Dataset

const ds = await client.openTable(tableName: string, readOnly?: boolean);

适用场景： 训练循环、批次迭代、PyTorch/TensorFlow DataLoader、异步预取。

对于只读操作，请传入 read_only=True。 默认以 读写模式打开，这会在数据集上获取写入句柄，并留下一个多余的 append / commit，可能会修改你本来只想检查的表。使用 read_only=True 时，你会获得一个 ReadOnlyDataset，其修改方法会报错。 读取、查询、summary()、批次迭代和 DataLoader 均可照常使用。

# Inspect without taking write access
ds = client.open_table("videos", read_only=True)
ds.summary()

# Batch iteration (training reads only — open read-only)
ds = client.open_table("videos", read_only=True)
for batch in ds.batches(32):
    train(batch)

# PyTorch DataLoader
from torch.utils.data import DataLoader
ds = client.open_table("training_data", read_only=True)
loader = DataLoader(ds.pytorch(), batch_size=32, shuffle=True, num_workers=4)

# TensorFlow tf.data.Dataset
ds = client.open_table("training_data", read_only=True)
tf_ds = ds.tensorflow().batch(32).prefetch(tf.data.AUTOTUNE)

# Writing still needs the default read/write handle
ds = client.open_table("training_data")
ds.append({"id": ["x"]})
ds.commit()

Node.js 说明： openTable(name, true) 需要包含 deeplake_open_read_only 绑定的 WASM 构建版本，而 describeTable() 需要包含 Dataset.summary() 的版本。较旧的构建包会清晰地指出缺少的绑定； 请重新构建 WASM 模块（python3 scripts/build_wasm_node.py dev）， 或改用 Python 端，该端对这两项功能的支持已推出更久。

---

查询

Fluent Query API（推荐）

client.table(table) 会返回一个可链式调用的 ManagedQueryBuilder：

# Python: supports both .execute() and () to run the query
results = (
    client.table("videos")
        .select("id", "text", "start_time")
        .where("file_id = $1", "abc123")
        .where("start_time > $2", 60)
        .order_by("start_time ASC")
        .limit(10)
        .offset(0)
)()  # or .execute()

// Node.js: use .execute() only (no () shorthand)
// (assumes initializeWasm() already called at startup)
const results = await client.table("videos")
    .select("id", "text", "start_time")
    .where("file_id = $1", "abc123")
    .where("start_time > $2", 60)
    .orderBy("start_time ASC")
    .limit(10)
    .offset(0)
    .execute();

方法	Python	Node.js	说明
.select(*cols)	.select("id", "t")	.select("id", "t")	设置列（默认值为 *）
.where(cond, *params)	.where("id=$1","x")	.where("id=$1","x")	添加 WHERE（多个条件以 AND 连接）
.order_by(clause)	.order_by("col")	.orderBy("col")	添加 ORDER BY
.limit(n)	.limit(10)	.limit(10)	设置 LIMIT
.offset(n)	.offset(20)	.offset(20)	设置 OFFSET
执行查询	.execute() 或 ()	.execute()	执行并返回结果

.where() 参数的工作方式： 每次调用 .where("col = $N", value) 都会添加一个 AND 条件。占位符 $1、$2 等由额外参数填充，并在所有 .where() 调用中按顺序编号。

原始 SQL：client.query()

# Python
rows = client.query(
    sql: str,
    params: tuple = None,
    timeout: int = 60,       # HTTP timeout in seconds (increase for slow queries)
) -> list[dict]

# Examples
rows = client.query("SELECT * FROM videos LIMIT 10")
rows = client.query("SELECT * FROM documents WHERE file_id = $1", ("abc123",))
rows = client.query("SELECT COUNT(*) FROM big_table", timeout=300)  # 5-minute timeout

// Node.js
const rows = await client.query(
    sql: string,
    params?: unknown[],
    options?: { timeoutMs?: number },  // default 60000 (60s)
) -> Promise<QueryRow[]>

// Examples
const rows = await client.query("SELECT * FROM videos LIMIT 10");
const rows = await client.query("SELECT * FROM documents WHERE file_id = $1", ["abc123"]);
const rows = await client.query("SELECT COUNT(*) FROM big", undefined, { timeoutMs: 300_000 });

查询通过 POST /workspaces/{id}/tables/query 发送到 API。参数化查询请使用 $1、$2 等。

超时： 默认查询超时时间为 60 秒。对于长时间运行的查询（大型聚合、索引构建），请通过 timeout / timeoutMs 参数延长超时时间。非默认超时时间会以 timeout_ms 的形式转发给后端，以便服务端也能应用截止时间。

有关 pg_deeplake SQL 功能（向量搜索、BM25、混合搜索、索引），请参阅 reference.md。

读己之写：摄取 → 查询的可见性延迟

ingest() / INSERT 成功后，写入不会立即对后续读取可见。写入会异步刷新到对象存储，因此在提交后几毫秒执行的 COUNT(*) 可能返回旧计数（新表通常为 0）——这是时序延迟，而非数据丢失。将摄取操作与查询串联时，可以：

• 轮询 SELECT COUNT(*) FROM <table>，直到计数达到预期值（请等待几秒——根据实测约为 5 秒）；或
• 如果有辅助进程在带外修改了同一数据集路径，随后读取又因 schema drift / 过期结果而失败，则清除缓存的快照：

SELECT deeplake_resync_table_pointer('my_table'::regclass);  -- re-open at HEAD
SELECT COUNT(*) FROM my_table;                                -- now current

deeplake_resync_table_pointer（pg_deeplake 1.13+）在存在尚未刷新的 DML 时会拒绝运行——请先调用 deeplake_flush_table('my_table'::regclass)。请参阅 reference.md。

SQL 方言 — Postgres wire，DuckDB 执行

针对 deeplake 表的查询会按 PostgreSQL 解析，但由底层的 DuckDB 引擎执行，因此两种方言并不完全兼容，差异偶尔会体现在错误消息中：

• jsonb 函数： DuckDB 提供的是 json_*，而不是 PG 的 jsonb_* 别名。jsonb_array_length(col) 会失败（“你是否想用 json_array_length”）；即便是 json_array_length(jsonb_col) 也可能失败。可靠的写法是 json_array_length(col::json)。建议使用 json_* 名称，并将 jsonb 转为 json。
• JSON 访问运算符： PG 的 -> / ->>'key' 运算符可以使用，但底层会映射到 DuckDB 路径语法（'$.key'）——这也是你会在错误消息中看到的内容。使用 json_extract(col, '$.key') 提取可避免意外。
• DDL 限制： 不支持原位迁移列类型——ALTER TABLE … ALTER COLUMN … TYPE … 会失败，并提示“VACUUM FULL is not supported for deeplake tables”。若要更改列类型，请使用目标架构创建新表并复制数据（INSERT INTO new SELECT … FROM old）。

---

表管理

# Python
tables = client.list_tables() -> list[str]
client.drop_table(table_name: str, if_exists: bool = True) -> None
client.create_table(table_name: str, schema: dict, *,
                    index=None, if_not_exists: bool = True) -> dict
client.create_index(table_name: str, column: str, *,
                    index_type: str = None, dimension: int = None) -> None
client.drop_index(table_name: str, column: str, if_exists: bool = True) -> None
client.describe_table(table_name: str) -> dict

// Node.js
const tables = await client.listTables();
await client.dropTable(tableName: string, ifExists?: boolean); // default true
await client.createTable(tableName: string, schema: SchemaMap,
                         options?: { index?: IndexSpec, ifNotExists?: boolean });
await client.createIndex(tableName: string, column: string,
                         options?: { indexType?: string, dimension?: number });
await client.dropIndex(tableName: string, column: string, ifExists?: boolean);
await client.describeTable(tableName: string);

创建表

ingest() 会在写入行时顺带创建表。create_table() / createTable() 则会使用显式架构创建一个空表。当表必须在数据写入之前就存在时，可使用此方法：例如，构建与现有数据集相对应的骨架表、为 INSERT INTO … SELECT 创建目标表，或预先声明无法根据值推断出的领域类型。

result = client.create_table("docs", {
    "id": "TEXT",
    "chunk_id": "TEXT",
    "created_at": "INT64",
    "embedding": "EMBEDDING(1536)",
}, index={
    "id": "inverted_index",
    "chunk_id": "inverted_index",
    "embedding": "clustered",
})
# {"table_name": "docs", "created": True, "dataset_path": "al://ws/docs"}

⚠️ 请写 EMBEDDING(dim)，不要只写 EMBEDDING。 单独的 EMBEDDING 会映射为无参数的 float4[]，pg_deeplake 会将其存储为可变长度浮点数组，而不是 embedding(dim, float32) 列。对于空骨架表，没有可用于推断维度的行，因此后续调用 create_index 时会缺少可用信息（必须手动传入 dimension=）。使用 EMBEDDING(1536) 时，该列会成为 embedding(1536, clustered)，与实际向量数据集一致。

当前，映射表无法识别的类型名称会原样传递给 PostgreSQL，因此 SMALLINT、TEXT[] 或 LINK(IMAGE) 都可以使用；拼写错误会作为服务器错误显现，而不会被悄悄转换为 TEXT 列。

EMBEDDING(N) 已自带 clustered 索引。 只需声明该列即可——在调用 create_index 之前，describe_table 就会报告 embedding(N, clustered)，并显示 index=clustered。因此，传入 index={"col": "clustered"} 是多余的（无害——类型相同，所以不会执行任何操作）；但如果在该列上请求不同的类型，则会失败：

Conflicting index types for column 'emb': existing index type 'clustered'

若要构建 clustered_quantized 索引，请将该列声明为纯 "EMBEDDING"（普通 float4[]，不自带索引），并在创建索引时传入维度：

client.create_table("vecs", {"emb": "EMBEDDING"})
client.create_index("vecs", "emb",
                    index_type="clustered_quantized", dimension=1536)

如果表已存在，create_table 会返回 created=False（传入 if_not_exists=False 则会改为报错）。与 ingest 不同，此处索引创建失败会抛出错误——没有已提交的数据需要保护。

创建索引

create_index() / createIndex() 会在列上创建 deeplake_index。使用它的场景包括：

• EMBEDDING 列——支持通过 <#> 执行向量余弦相似度搜索
• TEXT 列——支持 BM25 排序或关键词 CONTAINS 查询

该方法会执行 CREATE INDEX IF NOT EXISTS ... USING deeplake_index (column) WITH (index_type = '…')。

索引类型——以及为什么默认值通常不适合克隆数据集

列类型	接受的 index_type	省略时的默认值
文本	exact_text、bm25、inverted_index	exact_text
嵌入（1-D）	clustered、clustered_quantized	clustered
嵌入（2-D）	pooled_quantized	pooled_quantized
数值	inverted_index	inverted_index
json / dict	inverted_index	inverted_index
任意	unique	—

接受的别名：inverted → inverted_index、exact → exact_text、 quantized / binary_quantized → clustered_quantized。

⚠️ 文本类型的默认值与大多数真实数据集不符。 直接调用 create_index(table, "text") 会创建 exact_text，仅支持 col = 'x' 等值查询。源数据集通常使用 inverted_index（关键词 CONTAINS） 或 bm25（全文排序）。要克隆数据集的索引，必须显式传入 类型——应从 describe_table() 中读取，而不是猜测。

# Python — standalone
client.create_index("documents", "text", index_type="bm25")       # full-text ranking
client.create_index("documents", "keywords", index_type="inverted_index")
client.create_index("embeddings", "embedding", index_type="clustered")

# A clustered index over a bare float4[] column that has no rows yet needs the
# dimension supplied — there is nothing to infer it from.
client.create_index("skeleton", "embedding",
                    index_type="clustered", dimension=1536)

# Python — during ingestion (creates indexes after data is committed)
client.ingest("search_index", {
    "text": documents,
    "embedding": embeddings,
}, index={"text": "bm25", "embedding": "clustered"})

// Node.js — standalone
await client.createIndex("documents", "text", { indexType: "bm25" });
await client.createIndex("embeddings", "embedding", { indexType: "clustered" });
await client.createIndex("skeleton", "embedding",
                         { indexType: "clustered", dimension: 1536 });

// Node.js — during ingestion
await client.ingest("search_index", {
    text: documents,
    embedding: embeddings,
}, { index: { text: "bm25", embedding: "clustered" } });

index= 参数在 ingest() 和 create_table() 中均接受三种形式：

形式	含义
["embedding", "text"]	根据列类型使用默认索引类型
{"embedding": "clustered", "text": "bm25"}	为每列指定索引类型
{"embedding": {"index_type": "clustered", "dimension": 1536}}	完全自定义

更改现有索引类型

pg_deeplake 每列只允许一个索引。对于已建立索引的列，PostgreSQL 会跳过 CREATE INDEX IF NOT EXISTS，pg_deeplake 根本不会收到该操作——因此使用修正后的 index_type 重新运行也会悄无声息地不起作用。客户端会检查这种情况并报错：

client.create_index("docs", "text", index_type="bm25")
# TableError: Column 'text' of table 'docs' already has a 'exact_text' index
# (idx_docs_text), which does not match the requested 'bm25'. CREATE INDEX
# would skip it silently. Resolution: client.drop_index('docs', 'text') first.

client.drop_index("docs", "text")
client.create_index("docs", "text", index_type="bm25")   # now takes effect

重新请求相同类型时不会执行任何操作，因此可以安全地重新运行 create_table(..., index=...)。

⚠️ 目前在托管服务上，更改类型尚不能可靠生效。 DROP INDEX 会立即删除 PostgreSQL 目录项，但每个后端 Pod 都持有自己的 deeplake 索引副本，并会按各自的时间安排删除。之后立即使用新类型调用 create_index，尚未同步的 Pod 会拒绝请求，提示“已存在”或类型冲突—— 等待时间往往超过客户端约 10 秒的重试时限。根据 api-beta 的实测， 每 5 次运行中大约有 1 次成功。

客户端会处理短暂的延迟；尝试次数耗尽时，会报错，而不是报告一个实际并未创建的索引—— 因此你不会误以为某列已建立索引。遇到这种情况时，请等待后重试， 或从一开始就使用所需的索引类型创建表。请使用 describe_table() 验证，该方法读取的是存储，而不是目录。

读取表的实际架构和索引

describe_table() 以只读方式打开表，并报告存储中实际保存的内容，而不是创建表时的请求内容。内置的索引类型不会在 Python/JS API 的其他位置公开（它会被拼接到 Dataset.summary() 打印的类型字符串中），因此这是唯一的读取方式。

client.describe_table("embedding_noquantized")
# {'chunk_id':   {'type': 'text (compression=lz4)',      'index': 'inverted_index'},
#  'created_at': {'type': 'int64',                       'index': None},
#  'embedding':  {'type': 'embedding(1536, clustered)',  'index': 'clustered'},
#  'id':         {'type': 'text (compression=lz4)',      'index': 'inverted_index'}}

将其直接传回 create_table 即可克隆架构——完整的基本方案请参见 examples.md。

注意事项：克隆会复现列类型、维度和索引类型，但不会复现分块压缩——pg_deeplake 会将 PG TEXT 列硬编码为 text(compression=null)，因此，SDK 创建的源表显示 compression=lz4 时，此处会有所不同。差异仅影响存储大小；语义和查询不受影响。

**注意：**新创建的表需要一点时间才能在每个后端 pod 上可见。create_index 会自动处理这段时间（针对 "does not have DeepLake access method" / "does not exist" 进行有限次数的重试），因此可以一次性调用 create_table(..., index=...)。

---

工作区管理

工作区通过 REST API 创建。SDK 客户端没有内置的 createWorkspace() 方法——请直接使用 apiRequest。

**重要：**创建工作区时必须提供 id 字段。省略该字段会返回错误。

⚠️ 工作区会创建到哪个组织？POST /workspaces 会在请求所属范围的组织中创建工作区。该范围由 X-Activeloop-Org-Id 请求头决定——且只有该请求头能够指定范围。如果你属于多个组织，却没有传入组织请求头（或传入了拼写错误的请求头，例如 X-Org-Id），工作区会在令牌中嵌入的组织中静默创建，而不是你想要的组织——不会报错。如果目标组织与令牌所属组织不同，务必设置 X-Activeloop-Org-Id: <target-org-uuid>（或使用 org_id= 构造客户端，该参数会为你添加此请求头）。摄取数据前，请通过 GET /workspaces 检查（每一行的 org_id 字段）。

// Node.js — create workspace via API
import { apiRequest, extractOrgId } from 'deeplake';

// extractOrgId reads the token's embedded org. Pass a DIFFERENT org UUID here to
// target another org — apiRequest sends it as X-Activeloop-Org-Id.
const orgId = targetOrgId ?? extractOrgId(token);
await apiRequest(apiUrl, token, orgId, {
    method: 'POST',
    path: '/workspaces',
    body: { id: 'my-workspace', name: 'My Workspace' },
    timeoutMs: 30_000,
});

# Python — create workspace via API
import requests

resp = requests.post(
    f"{api_url}/workspaces",
    headers={"Authorization": f"Bearer {token}"},
    json={"id": "my-workspace", "name": "My Workspace"},
)
resp.raise_for_status()

列出工作区：

// Node.js
const resp = await apiRequest(apiUrl, token, orgId, {
    method: 'GET',
    path: '/workspaces',
});
// resp.data = [{ id, org_id, name, type, created_at }, ...]

字段	必填	说明
id	是	工作区标识符（用于 API 路径和 al:// URL）
name	否	显示名称（省略时默认为 id）

**注意：**省略 id 会返回 HTTP 400 Bad Request，消息为 "workspace ID is required"。

---

错误处理

Python 和 Node.js 使用相同的错误层次结构：

ManagedServiceError          # Base class for all errors
├── AuthError                # Token invalid/expired
│   └── TokenError           # Token parsing failed
├── CredentialsError         # DB credentials fetch failed
├── IngestError              # File ingestion failed
├── TableError               # Table operation failed
└── WorkspaceError           # Workspace not found or inaccessible

# Python imports
from deeplake.managed import (
    ManagedServiceError, AuthError, CredentialsError,
    IngestError, TableError, TokenError, WorkspaceError,
)

// Node.js imports
import {
    ManagedServiceError, AuthError, CredentialsError,
    IngestError, TableError, TokenError, WorkspaceError,
} from 'deeplake';

错误	原因	解决方案
AuthError: Token required	未提供令牌	向 Client() 传入 token=，或设置 DEEPLAKE_API_KEY 环境变量
AuthError: Token does not contain org_id	令牌缺少 OrgID 声明	确保令牌包含 OrgID 声明，或 API /me 可访问
IngestError: File not found	文件路径无效	检查指定路径下是否存在文件
TableError: table creation failed	API 表创建失败	检查 API 服务是否正在运行，以及工作区是否可访问
WorkspaceError: No storage path	API 未返回路径	检查工作区是否存在且已配置存储

故障排除详情请参见 reference.md。

---

可靠性与重试

托管后端是一个共享的自动扩缩容池。各个 pod 会重启、在负载增加时扩容，并刷新存储凭据——因此，瞬时错误很常见，稳健的客户端必须重试。请将这些行为纳入任何批量复制 / ETL 循环中。

重试临时 HTTP 状态码（采用退避策略）

状态码	含义	客户端操作
503（通常带有 Retry-After）	后端正在扩容／负载过高／连接失败——查询尚未运行	遵循 Retry-After（默认约 2s）；采用指数退避重试
504	查询超出截止时间	重试，或提高 timeout／添加 LIMIT
429	请求频率受限	遵循 Retry-After
查询返回 502 / 500	应该很少见；视为临时错误，并在有限次数内重试	采用退避策略重试；若问题持续，则发出警报

读取操作始终可以安全重试。对于写入操作，仅当写入具有幂等性时才重试（使用确定性主键 + ON CONFLICT DO NOTHING，或进行去重处理）——503 表示写入没有运行，但如果服务器已提交后连接才中断，则无法确定写入是否成功，因此默认采用幂等方式更安全。

⚠️ 静默截断：空页面并不总是表示数据已结束

对于分页客户端（LIMIT/OFFSET 或键集循环），最危险的故障模式是：后端凭据刷新期间，即使还有更多数据，页面也可能返回 HTTP 200 且行数为零。如果简单地把“空页面 ⇒ 完成”作为判断条件，就会静默截断——例如，只处理 249k 行中的 137k 行，却报告成功。请采取以下防护措施：

• 不要根据单个空页面推断数据已全部读取。 先交叉核对总数（SELECT COUNT(*)），并持续分页，直到读取的行数达到该总数。
• 优先使用键集分页（WHERE id > $last ORDER BY id LIMIT n），而不是 OFFSET，并确认最后一个键值持续递增。
• 如果数据流中途出现意外空页面，请在短暂退避后重新请求该页面，再判断是否已完成——真正的数据末尾页面重试后仍为空；凭据刷新导致的短暂问题则会在重试时恢复数据。
• 复制完成后，断言 rows_copied == COUNT(*)，若不匹配则明确报错。

写入后，验证已提交的行数

请参阅数据写入——ingest() 返回并不能证明事务已提交。将返回的 row_count 与发送的行数进行比较，并使用最终的 COUNT(*) 核对批量加载结果（同时考虑上述可见性延迟）。

---

Agent 决策树

决策：如何初始化客户端

Need to create a Client
|
|-- Python?
|   |-- DEEPLAKE_API_KEY env var is set?
|   |   `-- client = Client()                          # defaults: token from env, workspace="default"
|   |-- Have explicit token?
|   |   `-- client = Client(token="dl_xxx")            # workspace defaults to "default"
|   |-- Need specific workspace?
|   |   `-- client = Client(workspace_id="my-ws")      # token from env
|   `-- Need custom API URL?
|       `-- client = Client(api_url="http://custom:8080")
|
`-- Node.js?
    `-- import { ManagedClient, initializeWasm } from 'deeplake';
        await initializeWasm();
        const client = new ManagedClient({
           token: process.env.DEEPLAKE_API_KEY!,
           workspaceId: "my-ws",        // optional, default "default"
           apiUrl: "http://custom:8080", // optional
        });

决策：如何写入数据

User wants to ingest data
|
|-- Is it local files? -> use FILE schema
|   |-- Python:
|   |   `-- client.ingest("table", {"path": ["f1.mp4", "f2.mp4"]},
|   |          schema={"path": "FILE"})
|   `-- Node.js:
|       `-- await client.ingest("table", { path: ["f1.mp4"] },
|              { schema: { path: "FILE" } })
|
|-- Is it structured data (dict/lists)? -> pass a dict directly
|   |-- Python:
|   |   `-- client.ingest("table", {"col1": [...], "col2": [...]})
|   `-- Node.js:
|       `-- await client.ingest("table", { col1: [...], col2: [...] })
|
|-- Is it a HuggingFace dataset? -> use _huggingface key
|   `-- client.ingest("table", {"_huggingface": "dataset_name"})
|
|-- Is it a LeRobot robotics dataset? -> 3-table design (tasks + frames + episodes)
|   |-- from deeplake.managed.formats import LeRobotTasks, LeRobotFrames, LeRobotEpisodes
|   |-- client.ingest("tasks", format=LeRobotTasks(dataset_dir))
|   |-- client.ingest("frames", format=LeRobotFrames(dataset_dir, chunk_start=0, chunk_end=3))
|   `-- client.ingest("episodes", format=LeRobotEpisodes(dataset_dir, chunk_start=0, chunk_end=3))
|   Note: chunk_end is inclusive. Episodes requires git lfs. See examples.md Workflow 7.
|
|-- Is it a domain-specific format (COCO, etc.)? -> use format object
|   |-- Python:  client.ingest("table", format=my_format)
|   `-- Node.js: await client.ingest("table", null, { format: myFormat })
|   See formats.md for built-in formats: CocoPanoptic, Coco, LeRobot, custom
|
|-- Need custom chunking for text?
|   `-- client.ingest("table", {"path": ["doc.txt"]},
|          schema={"path": "FILE"},
|          chunk_size=500, chunk_overlap=100)
|
`-- Need explicit schema?
|   `-- client.ingest("table", {...}, schema={
|          "name": "TEXT",
|          "count": "INT64",
|          "vector": "EMBEDDING",
|      })
|
|
|-- Need the table to exist BEFORE any data? (skeleton, INSERT…SELECT target)
|   `-- client.create_table("table", {"id": "TEXT", "embedding": "EMBEDDING(1536)"},
|          index={"id": "inverted_index", "embedding": "clustered"})
|      Use EMBEDDING(dim), not bare EMBEDDING — an empty table has no rows to
|      infer the dimension from.
|
`-- Need indexes for search performance?
    `-- client.ingest("table", {...}, index=["embedding", "text"])
       Explicit types:  index={"text": "bm25", "embedding": "clustered"}
       Or standalone:   client.create_index("table", "embedding",
                                            index_type="clustered")
       The text default is exact_text (equality only) — pass bm25 or
       inverted_index if that is what you actually want.

决策：如何复现其他数据集的架构

Need a table with the same columns AND indexes as an existing dataset
|
|-- 1. Read the source (read-only — never take write access to inspect)
|      src = client.describe_table("source_table")
|      -> {col: {"type": "embedding(1536, clustered)", "index": "clustered"}}
|
|-- 2. Translate deeplake types -> schema types
|      "embedding(1536, ...)"  -> "EMBEDDING(1536)"   # keep the dimension!
|      "text (compression=…)"  -> "TEXT"
|      "int64"                 -> "INT64"
|
|-- 3. Carry each column's index type across verbatim
|      index = {c: i["index"] for c, i in src.items() if i["index"]}
|
`-- 4. client.create_table("skeleton", schema, index=index)
       Verify:  client.describe_table("skeleton") == src
       See examples.md Workflow 10 for the complete script.

决策：如何查询数据

User wants to query data
|
|-- Simple SELECT (small result)?
|   |-- Python fluent: client.table("table").select("id", "text").limit(100)()
|   |-- Python raw:    client.query("SELECT * FROM table LIMIT 100")
|   |-- Node fluent:   await client.table("table").select("id", "text").limit(100).execute()
|   `-- Node raw:      await client.query("SELECT * FROM table LIMIT 100")
|
|-- Large result set (streaming)?
|   `-- Use client.open_table("table", read_only=True) for direct dataset access
|      with batch iteration, PyTorch/TF DataLoaders, etc.
|
|-- Need semantic/vector search?
|   `-- from deeplake.managed import vector_literal
|      client.query(f"""
|          SELECT *, embedding::float4[] <#> {vector_literal(query_embedding)} AS score
|          FROM table ORDER BY score DESC LIMIT 10
|      """)
|      Through client.query() a vector CANNOT be passed as $1 -- the REST API
|      serializes it to a string that arrives at the DuckDB executor as
|      VARCHAR. Inline it. Scalars/text as $1 are fine. (On a direct psql /
|      asyncpg / psycopg connection $1::float4[] binds a real array and works.)
|
|-- Need text search?
|   `-- client.query("""
|          SELECT * FROM table
|          WHERE text @> $1
|      """, ("keyword",))
|
`-- Need hybrid search (vector + text)?
    `-- client.query(f"""
           SELECT *, (embedding, text)::deeplake_hybrid_record <#>
           deeplake_hybrid_record({vector_literal(query_emb)}, $1, 0.7, 0.3) AS score
           FROM table ORDER BY score DESC LIMIT 10
       """, ("search text",))
       Vector inlined; the text argument can stay a parameter.

决策：用户希望基于数据训练

User wants to train / iterate over data
|
|-- Fast native batch iteration (RECOMMENDED for large datasets)?
|   |-- ds = client.open_table("table", read_only=True)
|   |   for batch in ds.batches(256):  # dict of numpy arrays
|   |       states = torch.tensor(np.stack([batch[c] for c in cols], axis=1))
|   |-- For column subsets, use query first:
|   |   view = ds.query("SELECT col1, col2 WHERE episode_index < 100")
|   `   for batch in view.batches(256): ...
|
|-- Need PyTorch DataLoader (small datasets or need shuffle)?
|   `-- ds = client.open_table("table", read_only=True)
|      loader = DataLoader(ds.pytorch(), batch_size=32, shuffle=True)
|      NOTE: Slower on large remote datasets — prefer ds.batches() above
|
|-- Need TensorFlow tf.data?
|   `-- ds = client.open_table("table", read_only=True)
|      tf_ds = ds.tensorflow().batch(32).prefetch(AUTOTUNE)
|
`-- Training on LeRobot data?
    |-- Behavior cloning (state->action):
    |   ds = client.open_table("droid_frames", read_only=True)
    |   view = ds.query("SELECT state_x, ..., action_x, ... WHERE episode_index < 100")
    |   for batch in view.batches(256):
    |       states = torch.tensor(np.stack([batch[c] for c in STATE_COLS], axis=1))
    |       actions = torch.tensor(np.stack([batch[c] for c in ACTION_COLS], axis=1))
    |       # train(model, states, actions)
    |
    `-- Video-conditioned training:
        ds = client.open_table("droid_episodes", read_only=True)
        # Access video bytes via ds[i]["exterior_1_video"], etc.

决策：错误恢复

Operation failed with error
|
|-- AuthError?
|   |-- "Token required" -> Set DEEPLAKE_API_KEY env var or pass token= to Client()
|   |-- "Token does not contain org_id" -> Ensure token has OrgID claim
|   `-- "Token expired" -> Get new token
|
|-- IngestError?
|   |-- "data must be a dict" -> Pass a dict, not list/str/int
|   |-- "data must not be empty" -> Dict must have at least one key
|   |-- "File not found" -> Check file path exists
|   |-- "ffmpeg not found" -> Install ffmpeg for video processing
|   `-- "fitz not found" / "pdfjs-dist not found" -> Install pymupdf (Python) or pdfjs-dist (Node.js)
|
|-- TableError?
|   |-- "create_deeplake_table failed" -> Check pg_deeplake extension
|   |-- "Table already exists" -> Use drop_table() first or different name
|   |-- "Index creation failed" -> Check column exists and is EMBEDDING or TEXT type
|   |-- "already has index ... does not match the requested" ->
|   |      One index per column. client.drop_index(table, col), then recreate.
|   |-- "Invalid index type 'X' for text column" ->
|   |      Use exact_text | bm25 | inverted_index (embeddings: clustered |
|   |      clustered_quantized | pooled_quantized)
|   |-- "does not have DeepLake access method" right after create_table ->
|   |      Backend pod hasn't caught up. create_index already retries; if you
|   |      hit it elsewhere, wait a few seconds and retry.
|   |-- "not float32 array, hence not supported yet for indexing" ->
|   |      Column is not an embedding. Declare it EMBEDDING(dim), or pass
|   |      dimension= when indexing a bare float4[] column.
|   `-- "Query timed out" -> Increase timeout: client.query(sql, timeout=300)
|
|-- ValueError: "Unknown index_type"?
|   `-- Valid: inverted_index, bm25, exact_text, clustered,
|      clustered_quantized, pooled_quantized, unique
|
|-- Skeleton table's column came back as TEXT instead of a vector?
|   `-- Schema said "EMBEDDING" (no dimension). Use "EMBEDDING(1536)".
|
|-- ReadOnlyDatasetModificationError / AttributeError on append?
|   `-- The handle came from open_table(..., read_only=True). Reopen without it.
|
|-- WorkspaceError / "workspace ID is required"?
|   `-- POST /workspaces requires the "id" field in the request body.
|      Use: { id: "my-ws", name: "My Workspace" }
|
|-- "Not found: /workspaces/.../tables"?
|   `-- Workspace must exist before ingest(). Create it via the API or UI first.
|
|-- Tables created via raw SQL not visible to ingest()?
|   `-- Raw SQL tables (CREATE TABLE ... USING deeplake) are not registered with
|      the managed API. Use client.ingest() to create tables. Use client.query()
|      for raw SQL operations on manually-created tables.
|
|-- Thumbnail generation failed? (logged as warning, non-fatal)
|   |-- Python: Install Pillow (`pip install Pillow`)
|   `-- Node.js: Install sharp (`npm install sharp`)
|
`-- ManagedServiceError?
    `-- Check API server is running at the configured api_url

---

支持文件

• reference.md —— pg_deeplake SQL 参考（向量搜索、BM25、混合搜索、索引）、数据类型、LINK 类型与往返转换、附加现有 SDK 数据集（SECURITY LABEL physical_name + deeplake_resync_table_pointer）、托管凭据（Azure/GCS）与 deeplake_set_creds_key()、限制、性能调优、故障排查
• examples.md —— 完整的端到端工作流示例和详细的数据写入示例
• formats.md —— 格式基类、自定义格式类、normalize()/schema()/image_columns() 规则