一个专为AI设计的Python数据处理库：DataChain

贵阳哈哈镜

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：

1. {
   
2.     "class": "cat", "id": "1009", "num_annotators": 8,
   
3.     "inference": {"class": "dog", "confidence": 0.68}
   
4. }

复制代码

使用 JSON 元数据仅下载 “high-confidence cat” 推断图像的示例：

1. from datachain import Column, DataChain
   
2. 
   
3. meta = DataChain.from_json("gs://datachain-demo/dogs-and-cats/*json", object_name="meta")
   
4. images = DataChain.from_storage("gs://datachain-demo/dogs-and-cats/*jpg")
   
5. 
   
6. images_id = images.map(id=lambda file: file.path.split('.')[-2])
   
7. annotated = images_id.merge(meta, on="id", right_on="meta.id")
   
8. 
   
9. likely_cats = annotated.filter((Column("meta.inference.confidence") > 0.93) \
   
10.                                & (Column("meta.inference.class_") == "cat"))
    
11. likely_cats.export_files("high-confidence-cats/", signal="file")

复制代码

使用本地 AI 模型进行数据管理
使用 transformers 库通过简单的情感模型进行批量推理：pip install transformers

下面的代码将文件下载到云中，并将用户定义的函数应用于每个文件。然后，将检测到积极情绪的所有文件复制到本地目录。

1. from transformers import pipeline
   
2. from datachain import DataChain, Column
   
3. 
   
4. classifier = pipeline("sentiment-analysis", device="cpu",
   
5.                 model="distilbert/distilbert-base-uncased-finetuned-sst-2-english")
   
6. 
   
7. def is_positive_dialogue_ending(file) -> bool:
   
8.     dialogue_ending = file.read()[-512:]
   
9.     return classifier(dialogue_ending)[0]["label"] == "POSITIVE"
   
10. 
    
11. chain = (
    
12.    DataChain.from_storage("gs://datachain-demo/chatbot-KiT/",
    
13.                           object_name="file", type="text")
    
14.    .settings(parallel=8, cache=True)
    
15.    .map(is_positive=is_positive_dialogue_ending)
    
16.    .save("file_response")
    
17. )
    
18. 
    
19. positive_chain = chain.filter(Column("is_positive") == True)
    
20. positive_chain.export_files("./output")
    
21. 
    
22. print(f"{positive_chain.count()} files were exported")

复制代码

LLM 评审聊天机器人

LLMs 可以用作通用分类器。在下面的示例中，我们使用 Mistral 的免费 API 来判断公开可用的聊天机器人对话。请在 https://console.mistral.ai 获取免费的 Mistral API 密钥

1. $ pip install mistralai (Requires version >=1.0.0)
   
2. $ export MISTRAL_API_KEY=_your_key_

复制代码

DataChain 可以并行化 API 调用;免费的 Mistral 层级最多同时支持 4 个请求。

1. from mistralai import Mistral
   
2. from datachain import File, DataChain, Column
   
3. 
   
4. PROMPT = "Was this dialog successful? Answer in a single word: Success or Failure."
   
5. 
   
6. def eval_dialogue(file: File) -> bool:
   
7.      client = Mistral()
   
8.      response = client.chat.complete(
   
9.          model="open-mixtral-8x22b",
   
10.          messages=[{"role": "system", "content": PROMPT},
    
11.                    {"role": "user", "content": file.read()}])
    
12.      result = response.choices[0].message.content
    
13.      return result.lower().startswith("success")
    
14. 
    
15. chain = (
    
16.    DataChain.from_storage("gs://datachain-demo/chatbot-KiT/", object_name="file")
    
17.    .settings(parallel=4, cache=True)
    
18.    .map(is_success=eval_dialogue)
    
19.    .save("mistral_files")
    
20. )
    
21. 
    
22. successful_chain = chain.filter(Column("is_success") == True)
    
23. successful_chain.export_files("./output_mistral")
    
24. 
    
25. print(f"{successful_chain.count()} files were exported")

复制代码

更多案例可前往GitHub查看。

开源许可证

Apache-2.0 许可证