Inspect 数据集：CSV、JSON 和对话怎么读进评测

数据集

概述

Inspect 原生支持读取 CSV、JSON 和 JSON Lines 格式的数据集，也支持从 Hugging Face 读取。此外，评测流水线的核心数据集接口足够灵活，几乎可以接受从任何来源读取的数据（详见下文「自定义读取器」一节）。

如果你的数据已经是可以直接作为 Inspect 样本（一条交给模型的输入记录）读取的格式，读取数据集就这么简单：

from inspect_ai.dataset import csv_dataset, json_dataset
dataset1 = csv_dataset("dataset1.csv")
dataset2 = json_dataset("dataset2.json")

当然，许多真实世界的数据集并不会这么容易读取。下面我们将讨论把你的数据集适配到 Inspect 中使用的各种方式。

数据集样本

在 Inspect 中使用数据集所依据的核心数据类型是样本，它由一个必填的 input 字段以及若干其他可选字段组成：

类 inspect_ai.dataset.Sample

因此，具有如下结构的 CSV 数据集：

可以直接这样读取：

dataset = csv_dataset("security_guide.csv")

没有 id 字段的数据集，其样本会自动分配 id，依据是从 1 开始自增的整数。

如果你的样本包含 choices，则 target 应当是一个大写字母，表示 choices 中的正确答案，参见 multiple_choice。

样本文件

样本的 files 字段把沙箱中的目标文件路径映射到文件内容（内容可以是文件系统路径、URL，或包含内联内容的字符串）。例如，要把名为 flag.txt 的本地文件复制到沙箱路径 /shared/flag.txt，应这样写：

"/shared/flag.txt": "flag.txt"

文件会复制到默认沙箱环境，除非其名称包含前缀，从而映射到另一个环境。例如，要复制到 victim 沙箱：

"victim:/shared/flag.txt": "flag.txt"

你也可以指定一个目录而不是单个文件路径，它会被递归复制进沙箱：

"/shared/resources": "resources"

样本设置

setup 字段要么是 bash 设置脚本的路径（相对于数据集路径进行解析），要么是要执行的脚本内容。设置脚本以 5 分钟超时执行。如果你的设置脚本可能超过这个时长，应把一部分设置代码移到容器构建阶段的设置中（例如 Dockerfile）。

字段映射

如果你的数据集所含的输入与目标并不使用 input 和 target 作为字段名，可以用 FieldSpec 把它们映射到 Dataset 中。同一机制还使你能够把任意的附加字段收集到样本的 metadata 桶里。例如：

from inspect_ai.dataset import FieldSpec, json_dataset

dataset = json_dataset(
    "popularity.jsonl",
    FieldSpec(
        input="question",
        target="answer_matching_behavior",
        id="question_id",
        metadata=["label_confidence"],
    ),
)

要让 Sample.description（简短说明该样本要求智能体做什么）从某个数据集字段读取，用 FieldSpec(description=...) 指定该字段。除非指定了字段名，否则不会读取描述，因此已有的、带 description 列的数据集不受影响。该描述会随样本记录在评测日志中，并作为 samples_df() 的 description 列出现。

checkpoint 字段会被读取为该样本的检查点配置（Sample.checkpoint）。使用 FieldSpec(checkpoint=...) 可从另一个字段读取它；如果你的数据集有一个与此无关的 checkpoint 列，则使用 FieldSpec(checkpoint=None)。数据集记录不能设置检查点触发器；请在任务或 eval 上设置它。

如果你要做的不只是映射字段名，而是对数据做自定义处理，可以改为传入一个函数。该函数接收底层文件中的一条记录（表示为 dict），并返回一个样本。例如：

from inspect_ai.dataset import Sample, json_dataset

def record_to_sample(record):
    return Sample(
        input=record["question"],
        target=record["answer_matching_behavior"].strip(),
        id=record["question_id"],
        metadata={
            "label_confidence": record["label_confidence"]
        }
    )

dataset = json_dataset("popularity.jsonl", record_to_sample)

类型化元数据

如果你希望对样本元数据使用类型更强的接口，可以定义一个 Pydantic 模型，并用它来校验和读取元数据。

进行校验时，在 FieldSpec 中传入一个派生自 BaseModel 的类。元数据的接口是只读的，因此你还必须指定 frozen=True。例如：

from pydantic import BaseModel

class PopularityMetadata(BaseModel, frozen=True):
    category: str
    label_confidence: float

dataset = json_dataset(
    "popularity.jsonl",
    FieldSpec(
        input="question",
        target="answer_matching_behavior",
        id="question_id",
        metadata=PopularityMetadata,
    ),
)

要以类型安全的方式读取元数据，请在 Sample 或 TaskState 上使用 metadata_as() 方法：

metadata = state.metadata_as(PopularityMetadata)

再次注意，元数据的预期语义是只读的，因此尝试写入所返回的元数据会引发 Pydantic 的 FrozenInstanceError。

如果你需要每个样本各自可修改的数据，请使用 sample store，它同样支持使用 Pydantic 模型进行类型约束。

筛选

Dataset 类包含 filter() 与 shuffle() 方法，并支持切片运算符。

要选取数据集的一个子集，请使用 filter()：

dataset = json_dataset("popularity.jsonl", record_to_sample)
dataset = dataset.filter(
    lambda sample : sample.metadata["category"] == "advanced"
)

要选取记录的一个子集，请使用标准的 Python 切片：

dataset = dataset[0:100]

你也可以从命令行筛选，或在调用 eval() 时筛选。例如：

inspect eval ctf.py --sample-id 22
inspect eval ctf.py --sample-id 22,23,24
inspect eval ctf.py --sample-id *_advanced

上面最后一个例子演示了使用 glob（通配符）语法，用单个表达式选取多个样本。

打乱顺序

当你希望在评测开发过程中变换所使用的样本时，打乱顺序通常很有帮助。使用 --sample-shuffle 选项来执行打乱。例如：

inspect eval ctf.py --sample-shuffle
inspect eval ctf.py --sample-shuffle 42

也可以在 Python 中这样做：

eval("ctf.py", sample_shuffle=True)
eval("ctf.py", sample_shuffle=42)

你也可以直接在任务定义里打乱数据集。为此，可以使用 shuffle() 方法，或使用数据集加载函数的 shuffle 参数：

# shuffle method
dataset = dataset.shuffle()

# shuffle on load
dataset = json_dataset("data.jsonl", shuffle=True)

# shuffle on load with a seed
dataset = json_dataset("data.jsonl", shuffle=42)

注意，这两种方法都可选地支持为打乱指定随机种子。

选项打乱

在处理包含多项选择选项的数据集时，你可以在数据加载期间随机化这些选项的顺序。打乱操作会自动更新任何对应的 target 值，以维持正确答案的映射关系。

对于包含 choices 的数据集，你可以在加载数据时打乱选项。打乱选项会随机重排这些选项，并更新样本的一个或多个 target 值，使之与打乱后的选项对齐。

有两种打乱选项的方式：

# Method 1: Using the dataset method
dataset = dataset.shuffle_choices()

# Method 2: During dataset loading
dataset = json_dataset("data.jsonl", shuffle_choices=True)

若要得到可复现的打乱结果，可以指定随机种子：

# Using a seed with the dataset method
dataset = dataset.shuffle_choices(seed=42)

# Using a seed during loading
dataset = json_dataset("data.jsonl", shuffle_choices=42)

Hugging Face

Hugging Face Datasets 是一个便于访问和共享机器学习数据集的库，并与 Hugging Face Hub 集成。Hugging Face Hub 是一个收录了广泛公开共享数据集的仓库。通常，Hugging Face 上的数据集会要求指定使用其中的哪个划分（例如 train、test 或 validation），以及进行一些字段映射。使用 hf_dataset() 函数读取数据集，并指定所需的划分与字段名：

from inspect_ai.dataset import FieldSpec, hf_dataset

dataset=hf_dataset("openai_humaneval", 
  split="test", 
  sample_fields=FieldSpec(
    id="task_id",
    input="prompt",
    target="canonical_solution",
    metadata=["test", "entry_point"]
  )
)

注意，某些 HuggingFace 数据集会执行 Python 代码，以便解析底层的数据集文件。由于这些代码在你的本地机器上运行，你需要指定 trust = True 才能执行下载。只有对你所信任、并且你已经阅读过其代码的仓库，才应把该选项设为 True。下面是使用 trust 选项的例子（请注意，若未指定，它默认为 False）：

dataset=hf_dataset("openai_humaneval", 
  split="test", 
  trust=True,
  ...
)

在底层，hf_dataset() 函数调用的是 Hugging Face datasets 包中的 load_dataset() 函数。你还可以在调用 hf_dataset() 时附带任意参数，把它们继续传递给 load_dataset()。例如 hf_dataset(..., cache_dir="~/my-cache-dir")。

默认情况下，hf_dataset() 会以指数退避重试暂时性的 Hugging Face 错误（速率限制、超时，以及因 Hub 不可达而导致的缓存未命中）。传入 retry=False 即可禁用。

Amazon S3

Inspect 集成支持将数据集存储在 Amazon S3 上。与把数据存放在本地文件系统相比，使用 S3 可以提供更灵活的共享与访问控制，并且作为长期存储比本地文件更可靠。

使用 S3 基本上就是用 S3 URL（例如 s3://my-bucket-name）替换本地文件系统路径。例如，下面是从 S3 加载数据集的方法：

json_dataset("s3://my-bucket/dataset.jsonl")

S3 存储桶通常受到访问控制，因此读取需要身份验证。配置 AWS 身份验证客户端的方式多种多样，它们都能与 Inspect 配合使用。更多细节请参阅文章 Configuring the AWS CLI。

聊天消息

样本中最重要的数据结构是 ChatMessage。请注意，数据集的 input 常常是一个简单字符串（随后会在内部转换为 ChatMessageUser）。不过，也可以借助 ChatMessage 把完整的消息历史作为输入。ChatMessage 的另一个有用之处是提供多模态输入（例如图像）。

类 inspect_ai.model.ChatMessage

数据集中带有聊天消息的输入大致会像下面这样：

"input": [
  {
    "role": "user",
    "content": "What cookie attributes should I use for strong security?"
  }
]

请注意，对于这个例子，我们通常不会使用完整的聊天消息对象（而是只提供一个简单字符串）。当你希望加入系统提示，或用「assistant」回复来为对话做铺垫时，聊天消息对象更为有用。

自定义读取器

你并不限于使用内置的数据集函数来读取样本。你也可以构造一个 MemoryDataset，并把它传给任务。例如：

from inspect_ai import Task, task
from inspect_ai.dataset import MemoryDataset, Sample
from inspect_ai.scorer import model_graded_fact
from inspect_ai.solver import generate, system_message

dataset=MemoryDataset([
    Sample(
        input="What cookie attributes should I use for strong security?",
        target="secure samesite and httponly",
    )
])

@task
def security_guide():
    return Task(
        dataset=dataset,
        solver=[system_message(SYSTEM_MESSAGE), generate()],
        scorer=model_graded_fact(),
    )

因此，如果内置的数据集函数不能满足你的需求，你可以编写一个自定义函数来产出 MemoryDataset，并把它们直接传给你的 Task。

出处：UK AI Security Institute，Datasets，https://inspect.aisi.org.uk/datasets.html，MIT License。