项目介绍:DataChain
DataChain 是一个基于 Python 的人工智能数据仓库,专用于转换和分析非结构化数据,如图像、音频、视频、文本和 PDF。它能够与外部存储(如 S3)集成,有效处理数据而无需数据复制,并通过内部数据库管理元数据以实现高效查询。
应用场景
-
多模态数据集的准备与管理:DataChain 在数据的预训练、微调或大型语言模型(LLM)评估阶段非常理想,用于数据的组织和精炼。
-
生成式人工智能数据分析:支持多模态数据的高级分析,并利用大型语言模型进行临时分析。
关键特性
-
📂 多模态数据集版本控制:能够无冗余地版本化非结构化数据,支持引用 S3、GCP、Azure 和本地文件系统。支持多模态数据,包括图像、视频、文本、PDF、JSON、CSV、Parquet 等,将文件和元数据整合成持久的、版本化的列数据集。
-
🐍 Python 友好:可以操作 Python 对象和字段,处理浮点分数、字符串、矩阵以及 LLM 响应对象。支持在大规模数据集上运行 Python 代码,具备内建并行化和内存高效计算功能,无需使用 SQL 或 Spark。
-
🧠 数据丰富化与处理:利用本地 AI 模型和 LLM API 生成元数据,基于元数据进行过滤、连接和分组。支持向量嵌入搜索,并对 Python 对象进行高性能向量化操作。
快速开始
DataChain 的安装非常简单,仅需在终端中运行以下命令:
$ pip install datachain
使用JSON元数据选择文件
在一个包含猫和狗图像的存储中,每个图像都有相应的 JSON 文件,例如 cat.1009.json
。以下是利用 JSON 元数据仅下载“高置信度猫”图像的示例:
from datachain import Column, DataChain
meta = DataChain.from_json("gs://datachain-demo/dogs-and-cats/*json", object_name="meta")
images = DataChain.from_storage("gs://datachain-demo/dogs-and-cats/*jpg")
images_id = images.map(id=lambda file: file.path.split('.')[-2])
annotated = images_id.merge(meta, on="id", right_on="meta.id")
likely_cats = annotated.filter((Column("meta.inference.confidence") > 0.93) \
& (Column("meta.inference.class_") == "cat"))
likely_cats.export_files("high-confidence-cats/", signal="file")
本地AI模型的数据管理
使用 transformers
库进行批量推理,并将带有正面情感的文件复制到本地目录:
from transformers import pipeline
from datachain import DataChain, Column
classifier = pipeline("sentiment-analysis", device="cpu",
model="distilbert/distilbert-base-uncased-finetuned-sst-2-english")
def is_positive_dialogue_ending(file) -> bool:
dialogue_ending = file.read()[-512:]
return classifier(dialogue_ending)[0]["label"] == "POSITIVE"
chain = (
DataChain.from_storage("gs://datachain-demo/chatbot-KiT/",
object_name="file", type="text")
.settings(parallel=8, cache=True)
.map(is_positive=is_positive_dialogue_ending)
.save("file_response")
)
positive_chain = chain.filter(Column("is_positive") == True)
positive_chain.export_files("./output")
大型语言模型(LLM)评估
LLM 可用作通用分类器。以下示例展示了如何使用 Mistral API 评估聊天机器人对话:
from mistralai import Mistral
from datachain import File, DataChain, Column
PROMPT = "Was this dialog successful? Answer in a single word: Success or Failure."
def eval_dialogue(file: File) -> bool:
client = Mistral()
response = client.chat.complete(
model="open-mixtral-8x22b",
messages=[{"role": "system", "content": PROMPT},
{"role": "user", "content": file.read()}])
result = response.choices[0].message.content
return result.lower().startswith("success")
chain = (
DataChain.from_storage("gs://datachain-demo/chatbot-KiT/", object_name="file")
.settings(parallel=4, cache=True)
.map(is_success=eval_dialogue)
.save("mistral_files")
)
successful_chain = chain.filter(Column("is_success") == True)
successful_chain.export_files("./output_mistral")
print(f"{successful_chain.count()} files were exported")
DataChain 提供了强大的功能来处理和分析大规模和多样化的数据集,通过无缝集成和强大的处理能力,为数据科学家和工程师提供了一种高效且灵活的解决方案。