Python 实用工具：Activeloop 从入门到实战，轻松管理向量数据与大模型数据集

一、Activeloop 库简介

Activeloop 是一款专注于数据集管理、向量存储与大模型数据流水线的 Python 工具库，核心用于构建、版本控制、流式加载与查询 AI 数据集，尤其适配大语言模型、计算机视觉场景。其通过统一数据格式实现跨设备、跨框架的数据共享，底层依托高效序列化与云端存储能力，支持本地/云端无缝切换。该库采用 MIT License 开源，优点是易用、轻量、兼容主流 AI 框架，缺点是在超大规模离线数据集上性能略逊于专用分布式框架。

二、Activeloop 安装与环境准备

在使用 Activeloop 之前，我们需要完成库的安装与基础环境配置，支持 Python 3.8 及以上版本，可直接通过 pip 完成安装。

2.1 基础安装

打开命令行工具，执行以下安装命令：

pip install deeplake

Activeloop 的核心功能封装在 deeplake 包中，这是官方推荐的安装方式，安装过程会自动依赖 numpy、pandas 等基础数据处理库。

2.2 验证安装

安装完成后，可通过简单代码验证是否安装成功：

import deeplake

# 打印库版本
print("Activeloop (deeplake) 版本:", deeplake.__version__)

运行代码后，若正常输出版本号，说明环境配置完成，可进入后续功能使用。

三、Activeloop 核心功能与基础使用

Activeloop 核心围绕数据集创建、数据写入、数据读取、数据查询、向量存储五大功能展开，面向 AI 开发者屏蔽底层存储细节，专注数据本身。

3.1 创建本地数据集

数据集是 Activeloop 的核心载体，支持文本、图片、向量、标签等多种数据类型，创建方式简洁直观。

import deeplake
import numpy as np

# 创建本地数据集，路径为当前目录下的 my_first_dataset
ds = deeplake.dataset("./my_first_dataset")

# 定义数据集结构：文本数据、向量数据、标签
ds.create_tensor("text", htype="text")
ds.create_tensor("embedding", htype="embedding")
ds.create_tensor("label", htype="class_label")

print("数据集创建完成，结构如下：")
print(ds.tensors)

代码说明：

• deeplake.dataset() 用于创建/加载数据集，传入本地路径则生成本地数据集；
• create_tensor() 定义数据列，htype 指定数据类型，适配 AI 场景常用格式；
• 执行后会在指定路径生成数据集文件夹，包含数据与元信息。

3.2 向数据集写入数据

创建完数据集结构后，可批量写入文本、向量、标签等数据，模拟大模型训练或检索场景的原始数据。

# 构造模拟数据
text_list = [
    "Python 是一门优雅易用的编程语言",
    "Activeloop 适合管理大模型数据集",
    "向量数据库是 RAG 系统的核心组件",
    "深度学习需要高质量标注数据"
]

# 生成模拟 768 维嵌入向量（适配大模型通用维度）
embedding_list = [np.random.rand(768) for _ in range(4)]
label_list = [0, 1, 1, 0]

# 批量写入数据集
with ds:
    ds.text.extend(text_list)
    ds.embedding.extend(embedding_list)
    ds.label.extend(label_list)

print("数据写入成功，数据集样本数：", len(ds))

代码说明：

• 使用 with ds: 上下文管理器保证数据写入原子性，避免中途出错导致数据损坏；
• extend() 用于批量添加数据，适配列表、numpy 数组等格式；
• 写入后可通过 len(ds) 查看数据集总样本数。

3.3 读取数据集数据

Activeloop 支持索引读取、遍历读取、条件读取，操作方式与列表、DataFrame 高度相似，降低学习成本。

# 按索引读取单条数据
sample = ds[0]
print("第一条数据文本：", sample.text.data())
print("第一条数据向量形状：", sample.embedding.shape)
print("第一条数据标签：", sample.label.data())

# 遍历所有数据
print("\n===== 遍历全部数据 =====")
for i, sample in enumerate(ds):
    print(f"样本 {i}：")
    print(f"文本：{sample.text.data()}")
    print(f"标签：{sample.label.data()}\n")

代码说明：

• 直接通过索引 ds[i] 获取第 i 条样本；
• data() 方法提取原始数据，避免返回封装对象；
• 遍历方式与 Python 列表一致，无需复杂语法。

3.4 条件查询数据

Activeloop 内置轻量查询引擎，支持按标签、数值等条件筛选数据，满足 AI 数据预处理需求。

# 查询标签为 1 的所有样本
filtered_ds = ds[ds.label == 1]

print("标签为 1 的样本数量：", len(filtered_ds))
for sample in filtered_ds:
    print("筛选文本：", sample.text.data())

代码说明：

• 支持 ==、!=、>、< 等常规比较运算符；
• 筛选后返回新的数据集视图，不占用额外内存；
• 适合大模型训练前的数据过滤与采样。

四、Activeloop 向量存储与 RAG 场景实战

向量存储是 Activeloop 的核心亮点，可直接作为轻量级向量数据库使用，适配 RAG（检索增强生成）场景，无需额外部署复杂数据库。

4.1 向量数据写入与检索

# 重新创建向量专用数据集
rag_ds = deeplake.dataset("./rag_embedding_dataset")
rag_ds.create_tensor("content", htype="text")
rag_ds.create_tensor("vector", htype="embedding")

# 写入文档与对应向量
documents = [
    "豆包是字节跳动自研的人工智能助手",
    "Activeloop 可用于构建 RAG 系统的向量库",
    "Python 广泛应用于 AI 与数据科学领域",
    "RAG 系统通过检索提升大模型回答准确性"
]

# 生成模拟向量
vectors = [np.random.rand(128) for _ in range(4)]

with rag_ds:
    rag_ds.content.extend(documents)
    rag_ds.vector.extend(vectors)

# 模拟查询向量并计算相似度
query_vector = np.random.rand(128)
scores = []

for sample in rag_ds:
    vec = sample.vector.data()
    # 余弦相似度简化计算
    similarity = np.dot(query_vector, vec) / (np.linalg.norm(query_vector) * np.linalg.norm(vec))
    scores.append(similarity)

# 获取最相似的文档
best_idx = np.argmax(scores)
print("\n最匹配的文档：", rag_ds[best_idx].content.data())
print("匹配相似度：", round(scores[best_idx], 4))

代码说明：

• 该示例完整模拟 RAG 系统中文档入库→向量存储→相似度检索流程；
• 无需依赖 Pinecone、Chroma 等外部向量库，单机即可运行；
• 适合个人开发者、小型项目快速搭建检索系统。

4.2 与大模型嵌入接口结合

Activeloop 可无缝对接 OpenAI、文心一言、豆包等大模型的嵌入接口，实现真实向量生成与存储。

# 模拟调用大模型生成嵌入向量（可替换为真实 API）
def mock_get_embedding(text: str) -> np.ndarray:
    return np.random.rand(128)

# 构建真实场景数据集
qa_ds = deeplake.dataset("./qa_dataset")
qa_ds.create_tensor("question", htype="text")
qa_ds.create_tensor("answer", htype="text")
qa_ds.create_tensor("q_embedding", htype="embedding")

qa_pairs = [
    {"q": "Activeloop 是什么", "a": "Activeloop 是 Python 数据集与向量管理库"},
    {"q": "如何安装 deeplake", "a": "使用 pip install deeplake 安装"},
    {"q": "RAG 全称是什么", "a": "RAG 全称是 Retrieval Augmented Generation"}
]

with qa_ds:
    for pair in qa_pairs:
        q_emb = mock_get_embedding(pair["q"])
        qa_ds.question.append(pair["q"])
        qa_ds.answer.append(pair["a"])
        qa_ds.q_embedding.append(q_emb)

print("问答数据集构建完成，样本数：", len(qa_ds))

代码说明：

• 将 mock_get_embedding 替换为真实模型接口，即可生成工业级向量库；
• 数据集同时存储问题、答案、向量，形成完整 RAG 数据链路。

五、Activeloop 云端数据集使用

Activeloop 支持云端存储，实现多设备、多开发者共享数据集，无需手动传输文件。

5.1 登录与云端数据集创建

首先在命令行登录 Activeloop 账号：

deeplake login

按照提示输入用户名与密码，登录成功后即可创建云端数据集。

# 创建云端数据集（需登录）
# cloud_ds = deeplake.dataset("hub://用户名/my_cloud_dataset")

# 后续读写操作与本地数据集完全一致
# with cloud_ds:
#     cloud_ds.text.extend(["云端数据测试"])

代码说明：

• 路径以 hub:// 开头表示云端数据集；
• 读写 API 与本地完全一致，实现本地/云端无缝切换；
• 适合团队协作、跨设备开发。

六、与 PyTorch/TensorFlow 框架对接

Activeloop 原生支持深度学习框架，可直接转换为框架可读取的数据集，简化训练数据加载流程。

# 转换为 PyTorch DataLoader
from torch.utils.data import DataLoader

# 构建适合训练的数据集
train_ds = deeplake.dataset("./train_dataset")
train_ds.create_tensor("image", htype="image")
train_ds.create_tensor("target", htype="class_label")

# 模拟图像数据
for i in range(10):
    train_ds.image.append(np.random.rand(28, 28, 3))
    train_ds.target.append(np.random.randint(0, 2))

# 转换为 PyTorch 数据集
pytorch_ds = train_ds.pytorch(batch_size=2, shuffle=True)
dataloader = DataLoader(pytorch_ds, batch_size=None)

# 读取训练批次
for batch in dataloader:
    print("图像批次形状：", batch["image"].shape)
    print("标签批次：", batch["target"])
    break

代码说明：

• 通过 pytorch() 方法直接生成适配框架的数据格式；
• 支持 shuffle、batch_size、num_workers 等训练参数；
• 大幅减少数据预处理与格式转换代码量。

七、实际应用案例：简易智能问答系统

结合前面所有知识点，构建一个可直接运行的轻量级智能问答系统，完整体现 Activeloop 的实用价值。

import deeplake
import numpy as np

# 1. 构建知识库数据集
kb_ds = deeplake.dataset("./knowledge_base")
kb_ds.create_tensor("question", htype="text")
kb_ds.create_tensor("answer", htype="text")
kb_ds.create_tensor("embed", htype="embedding")

# 知识库内容
knowledge = [
    {"q": "Python 有哪些常用数据结构", "a": "列表、字典、元组、集合、堆、队列等"},
    {"q": "Activeloop 能做什么", "a": "管理数据集、存储向量、构建 RAG、对接深度学习框架"},
    {"q": "如何读取 deeplake 数据", "a": "通过索引 ds[i] 或遍历读取，使用 data() 获取原始数据"},
    {"q": "deeplake 支持云端吗", "a": "支持，使用 hub:// 路径创建云端数据集"}
]

# 写入数据
with kb_ds:
    for item in knowledge:
        embed = np.random.rand(128)
        kb_ds.question.append(item["q"])
        kb_ds.answer.append(item["a"])
        kb_ds.embed.append(embed)

# 2. 定义检索函数
def search_answer(user_query: str) -> str:
    query_embed = np.random.rand(128)
    max_sim = -1
    best_ans = ""

    for sample in kb_ds:
        sim = np.dot(query_embed, sample.embed.data()) / (
            np.linalg.norm(query_embed) * np.linalg.norm(sample.embed.data())
        )
        if sim > max_sim:
            max_sim = sim
            best_ans = sample.answer.data()
    return best_ans

# 3. 模拟用户提问
if __name__ == "__main__":
    user_query = "如何读取 deeplake 里面的数据"
    answer = search_answer(user_query)

    print("用户问题：", user_query)
    print("系统回答：", answer)

该案例可直接部署在本地，作为小型客服机器人、文档问答助手使用，代码简洁、依赖少、启动快，充分体现 Activeloop 在 AI 小项目中的高效性。

相关资源

• Pypi地址：https://pypi.org/project/deeplake/
• Github地址：https://github.com/activeloopai/deeplake
• 官方文档地址：https://docs.activeloop.ai/

关注我，每天分享一个实用的Python自动化工具。