DataChain 是一个现代化的 Python 数据框架库,专为人工智能而设计。它旨在将您的非结构化数据组织成数据集,并在本地机器上大规模处理。DataChain 不会抽象或隐藏 AI 模型和 API 调用,而是帮助将它们集成到后现代数据堆栈中。
GitHub：https://github.com/iterative/datachain
核心功能
将存储作为唯一数据源
支持多模态数据:图像、视频、文本、PDF、JSON、CSV、Parquet 等
将文件和元数据合并为持久、版本化、列式数据集
提供 Python 友好的数据管道
内置并行化和内存外计算,无需 SQL 或 Spark
使用本地 AI 模型和 LLM API 进行数据丰富和处理
提供高效的并行化、内存外工作负载和数据缓存
技术栈
Python
机器学习
数据处理和分析
使用案例
安装：
$ pip install datachain
使用 JSON 元数据选择文件
一个存储由猫和狗的图像（dog.1048.jpg、cat.1009.jpg）组成，并以“json-pairs”格式注释有地面实况和模型推理，其中每个图像都有一个匹配的 JSON 文件，如 cat.1009.json：
{
"class": "cat", "id": "1009", "num_annotators": 8,
"inference": {"class": "dog", "confidence": 0.68}
}
使用 JSON 元数据仅下载 “high-confidence cat” 推断图像的示例：
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('.'))
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 库通过简单的情感模型进行批量推理：pip install 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()
return classifier(dialogue_ending) == "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")
print(f"{positive_chain.count()} files were exported")
LLM 评审聊天机器人LLMs 可以用作通用分类器。在下面的示例中，我们使用 Mistral 的免费 API 来判断公开可用的聊天机器人对话。请在 https://console.mistral.ai 获取免费的 Mistral API 密钥$ pip install mistralai (Requires version >=1.0.0)
$ export MISTRAL_API_KEY=_your_key_DataChain 可以并行化 API 调用;免费的 Mistral 层级最多同时支持 4 个请求。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.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")更多案例可前往GitHub查看。开源许可证Apache-2.0 许可证