camel.datasets 包

camel.datasets 包#

子模块#

camel.datasets.base 模块#

模块内容#

class camel.datasets.BaseGenerator(seed: int = 42, buffer: int = 20, cache: str | Path | None = None, data_path: str | Path | None = None, **kwargs)[来源]#

基类: ABC, IterableDataset

数据生成器的抽象基类。

该类定义了生成合成数据点的接口。 具体实现应提供特定的生成策略。

async async_sample() → DataPoint[来源]#

从当前数据集中异步返回下一个数据点。

Returns:

下一个数据点。

Return type:

DataPoint

注意

此方法适用于异步上下文。在同步上下文中请使用'sample'。

flush(file_path: str | Path) → None[来源]#

将当前数据刷新到JSONL文件并清除数据。

Parameters:

file_path (Union[str, Path]) – 保存JSONL文件的路径。

注意事项

  • 使用save_to_jsonl保存self._data。

abstract async generate_new(n: int, **kwargs) → None[来源]#

生成n个新数据点并将它们附加到self._data。

子类实现必须生成指定数量的数据点，并将它们直接追加到self._data列表中。 此方法不应返回数据点；迭代器依赖于self._data被填充。

Parameters:

  • n (int) – 要生成并追加的数据点数量。

  • **kwargs – 额外的生成参数。

Returns:

此方法不应返回任何内容。

Return type:

无

示例

```python async def generate_new(self, n: int, **kwargs) -> None:

new_points = [DataPoint(…) for _ in range(n)] self._data.extend(new_points)

```

sample() → DataPoint[来源]#

从当前数据集中同步返回下一个数据点。

Raises:

RuntimeError - 如果在异步上下文中调用时抛出。

Returns:

下一个数据点。

Return type:

DataPoint

注意

该方法适用于同步上下文。 在异步上下文中请使用'async_sample'， 以避免阻塞或运行时错误。

save_to_jsonl(file_path: str | Path) → None[来源]#

将生成的数据点保存到JSONL（JSON行）文件中。

每个数据点都以单独的JSON对象形式存储在新的一行中。

Parameters:

file_path (Union[str, Path]) – 保存JSONL文件的路径。

Raises:

  • ValueError - 如果没有生成任何数据点。

  • IOError - 如果写入文件时出现问题。

注意事项

  • 使用self._data，其中包含生成的数据点。

  • 如果文件已存在，则追加内容。

  • 通过使用JSONL格式确保与大型数据集的兼容性。

class camel.datasets.DataPoint(*, question: str, final_answer: str, rationale: str | None = None, metadata: Dict[str, Any] | None = None)[来源]#

基类：BaseModel

数据集中的单个数据点。

question#

需要解决的主要问题或议题。

Type:

字符串

final_answer#

最终答案。

Type:

字符串

rationale#

答案背后的逻辑推理或解释。(默认: None)

Type:

可选[str]

metadata Optional[Dict[str, Any]]

关于数据点的额外元数据。(默认: None)

final_answer: str#
classmethod from_dict(data: Dict[str, Any]) → 数据点[来源]#

从字典创建一个DataPoint。

Parameters:

data (Dict[str, Any]) – 包含DataPoint字段的字典。

Returns:

新建DataPoint实例。

Return type:

DataPoint

metadata: Dict[str, Any] | None#
model_config: ClassVar[ConfigDict] = {}#

模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

question: str#
rationale: str | None#
to_dict() → Dict[str, Any][来源]#

将DataPoint转换为字典。

Returns:

DataPoint的字典表示形式。

Return type:

字典[字符串, 任意类型]

class camel.datasets.FewShotGenerator(seed_dataset: StaticDataset, verifier: BaseVerifier, model: BaseModelBackend, seed: int = 42, **kwargs)[来源]#

基类: BaseGenerator

一个使用少样本学习生成合成数据点的生成器。

该类利用种子数据集、代理和验证器，通过少量提示按需生成新的合成数据点。

async generate_new(n: int, max_retries: int = 10, num_examples: int = 3, **kwargs) → None[来源]#

通过少量样本提示生成并验证n个新数据点，设有重试次数限制。

Steps:

  1. 从种子数据集中采样示例。

  2. 使用选定的示例构建提示。

3. 使用代理生成新的数据点，包含问题及解决问题的代码。 4. 使用验证器执行代码以获取伪真实值。 5. 将有效数据点存储在内存中。

Parameters:

  • n (int) – 要生成的有效数据点数量。

  • max_retries (int) - 停止前的最大重试次数。 (默认值: 10)

  • num_examples (int) – 从数据中采样的示例数量

  • 提示. (用于少量样本的种子数据集) – (默认: 3)

  • **kwargs – 额外的生成参数。

Returns:

新生成的有效数据点列表。

Return type:

列表[DataPoint]

Raises:

  • TypeError - 如果代理的输出不是字典（或不符合预期格式）。

  • KeyError – 如果响应中缺少必需的键。

  • AttributeError - 如果验证器响应缺少属性。

  • ValidationError - 如果数据点未能通过模式验证。

  • RuntimeError – 如果在生成n个有效数据点之前重试次数已耗尽。

注意事项

  • 在验证失败时重试，直到存在n个有效数据点

    或达到max_retries上限，以先到者为准。

  • 如果在达到n之前重试次数耗尽，将抛出RuntimeError

    异常。

  • 元数据包含一个时间戳，用于追踪数据点的创建。

class camel.datasets.SelfInstructGenerator(seed_dataset: StaticDataset, verifier: BaseVerifier, instruction_agent: ChatAgent | None = None, rationale_agent: ChatAgent | None = None, seed: int = 42, **kwargs)[来源]#

基类: BaseGenerator

一个使用self-instruct创建合成数据点的生成器。

它同时利用人工提供的数据集(seed_dataset)和生成的机器指令(machine_instructions)来产生新的合成数据点，这些数据点包含一个问题、计算出的推理过程(代码)以及来自验证器的最终答案。

class QuestionSchema(*, question: str)[来源]#

基类：BaseModel

生成问题的模式。

question#

模型生成的问题。

Type:

字符串

model_config: ClassVar[ConfigDict] = {}#

模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

question: str#
class RationaleSchema(*, code: str)[来源]#

基类：BaseModel

生成原理代码的Schema。

code#

生成的代码未经任何格式化。

Type:

字符串

code: str#
model_config: ClassVar[ConfigDict] = {}#

模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

default_instruction_agent() → ChatAgent[来源]#

创建默认的指令生成代理。

该代理配置了适中的温度设置，以鼓励生成富有创意且多样化的指令行为。

Returns:

一个带有默认指令提示的代理。

Return type:

ChatAgent

default_rationale_agent() → ChatAgent[来源]#

创建默认的rationale生成代理。

该代理配置为确定性（零温度）设置，以确保基于给定指令和包列表生成一致且精确的推理。

Returns:

一个带有推理提示的代理

Return type:

ChatAgent

static format_support_block(dp: DataPoint) → str[来源]#

将DataPoint格式化为少量示例块。

Parameters:

dp (DataPoint) – 一个数据点。

Returns:

一个格式化字符串，包含问题及其

对应的Markdown风格Python格式代码块。

Return type:

字符串

async generate_new(n: int, max_retries: int = 10, human_sample_count: int = 3, machine_sample_count: int = 1, **kwargs) → None[来源]#

通过自我指导提示生成并验证n个新数据点，设有重试次数限制。

Parameters:

  • n (int) – 要生成的有效数据点数量。

  • max_retries (int) - 停止前的最大重试次数。 (默认值: 10)

  • human_sample_count (int) – 要采样的人类示例数量。 (默认值: 3)

  • machine_sample_count (int) – 要采样的机器示例数量。 (默认: 1)

  • **kwargs - 额外的关键字参数。

注意事项

  • 在验证失败时重试，直到存在n个有效数据点

    或达到max_retries上限，以先到者为准。

  • 如果在达到n之前重试次数耗尽，将抛出RuntimeError

    异常。

  • 元数据包含一个时间戳，用于追踪数据点的创建。

generate_new_instruction(agent: ChatAgent, support_human_dps: list[DataPoint], support_machine_dps: list[DataPoint]) → str[来源]#

使用自指导提示生成新指令。

Parameters:

  • agent (ChatAgent) – 用于生成指令的代理。

  • support_human_dps (list[DataPoint]) – 用于采样的人类示例列表。

  • support_machine_dps (list[DataPoint]) – 用于采样的机器示例列表。

Returns:

新生成的问题。

Return type:

字符串

generate_rationale(question: str, agent: ChatAgent | None = None, support_human_dps: list[DataPoint] | None = None) → str[来源]#

为给定问题生成合理的代码（解决方案）。

Parameters:

  • question (str) – 需要解决的问题。

  • agent (可选[ChatAgent]) – 用于生成原理的代理。如果未提供，将使用默认的原理代理。(默认: None)

  • support_human_dps (Optional[list[DataPoint]]) – 用于采样的人类示例列表。(默认: None)

Returns:

生成的代码解决方案以字符串形式呈现。

Return type:

字符串

class camel.datasets.StaticDataset(data: Dataset | Dataset | Path | List[Dict[str, Any]], seed: int = 42, min_samples: int = 1, strict: bool = False, **kwargs)[来源]#

基类：Dataset

一个包含数据点列表的静态数据集。 确保所有项目都符合DataPoint模式。 该数据集扩展自PyTorch的Dataset， 应在运行时大小固定的情况下使用。

该类可以从Hugging Face数据集、PyTorch数据集、JSON文件路径或字典列表初始化，并将它们转换为一致的内部格式。

property metadata: Dict[str, Any]#

检索数据集元数据。

Returns:

数据集元数据字典的副本。

Return type:

字典[字符串, 任意类型]

sample() → DataPoint[来源]#

从数据集中随机抽取一个数据点样本。

Returns:

一个随机抽样的DataPoint。

Return type:

DataPoint

Raises:

RuntimeError – 如果数据集为空且无法抽取样本时抛出。

On this page