camel.environments 包

本页内容

camel.environments 包#

子模块#

camel.environments.base 模块#

模块内容#

class camel.environments.Action(*, index: int = 0, llm_response: str, metadata: ~typing.Dict[str, ~typing.Any] = <factory>, timestamp: ~datetime.datetime = <factory>)[来源]#

基类：BaseModel

表示在环境中采取的一个动作。

这个类定义了输入上下文、LLM生成的输出以及在强化学习框架内进行验证和跟踪所需的元数据。

llm_response#

由LLM生成的响应。

Type:: 字符串

metadata#

额外的元数据，如模型参数、提示详情或响应置信度分数。

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

timestamp#

动作生成时的时间戳（UTC）。

Type:: 日期时间

index: int#

llm_response: str#

metadata: Dict[str, Any]#

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

timestamp: datetime#

class camel.environments.Environment(*args, **kwargs)[来源]#

基类: Protocol

async close() → None[来源]#: 对所有环境资源执行全面清理。

async reset() → 观察[来源]#

将环境重置为初始状态。

Returns:: 该剧集的初步观察

async step(action: 操作) → StepResult[来源]#

在环境中执行一步操作。

Parameters:

action – 包含所需一切内容的操作
environment (在...中取得进展)

Returns:

包含下一个观察结果、奖励、完成标志和信息的StepResult

class camel.environments.MultiStepEnv(extractor: BaseExtractor, max_steps: int | None = None, **kwargs)[来源]#

基类: ABC

一个用于强化学习与LLMs的多步骤环境。

async close() → None[来源]#

清理并关闭环境使用的所有资源。该方法会关闭验证器，调用任何MultiStepEnv中实现的内部关闭函数，并确保环境被正确关闭。

Raises:: 异常 – 如果在关闭环境时发生错误。

abstract async compute_reward() → Tuple[float, Dict[str, float]][来源]#

property current_step: int#

获取当前步骤编号。

Returns:: 当前所处的步骤编号。
Return type:: int

is_done() → bool[来源]#

检查该片段是否应该终止。

当达到最大步数或满足任何其他终止条件时，此函数将终止当前回合。

Returns:: 一个布尔标志。
Return type:: 布尔值

property metadata: Dict[str, Any]#

检索环境的元数据。

这提供了额外的参数和配置详情。

Returns:: 环境元数据的副本。
Return type:: 字典[字符串, 任意类型]

async reset() → 观察[来源]#

将环境重置为初始状态。

Returns:: 该剧集的初始观察。
Return type:: 观察
Raises:: RuntimeError – 如果无法获取初始观察值。

async setup() → None[来源]#

通过初始化验证器和提取器来设置环境。

该方法确保环境已准备好进行交互。它设置了必要的组件，包括验证器和提取器。

Raises:: 异常 - 如果由于内部错误导致设置失败。

async step(action: 操作) → Tuple[观察, float, bool, Dict[str, Any]][来源]#

使用给定的动作在环境中迈出一步。

该方法根据LLM的响应更新环境状态，计算奖励值，检查当前回合是否结束，并据此获取下一个或最终的观察结果。

Parameters:

action (Action) – 包含LLM响应的操作。

Returns:

包含下一个观察结果、总奖励、奖励字典的StepResult: 完成标志和信息。

Raises:

RuntimeError - 如果环境未设置、回合已结束或没有有效的当前观察值。

class camel.environments.Observation(*, question: str, context: ~typing.Dict[str, ~typing.Any] = <factory>, metadata: ~typing.Dict[str, ~typing.Any] | None = None)[来源]#

基类：BaseModel

环境观测。

question#

向LLM提出的问题。

Type:: 字符串

context#

问题的附加背景信息。

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

metadata#

关于观测的可选元数据。

Type:: 字典[str, 任意类型] | 无

context: Dict[str, Any]#

metadata: Dict[str, Any] | None#

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

question: str#

class camel.environments.Opponent(play_style: Literal['optimal', 'random'] = 'optimal')[来源]#

基类：object

Tic Tac Toe游戏的AI对手。

该类实现了AI对手的不同游戏策略，包括使用带有alpha-beta剪枝的minimax算法的最优策略，以及随机策略。

get_optimal_move(board: List[str]) → int | None[来源]#

使用极小化极大算法获取最优移动。

Parameters:

board (List[str]) – 当前游戏棋盘，以字符串列表形式表示。

Returns:

最佳移动的索引，如果没有可用移动则为None: 。

Return type:

可选[int]

minimax(board: List[str], is_maximizing: bool, depth: int = 0, alpha: float = -inf, beta: float = inf) → Tuple[float, int | None][来源]#

采用alpha-beta剪枝的Minimax算法用于最优走子选择。

递归评估所有可能的走法以找到最佳选择。使用alpha-beta剪枝算法来减少搜索空间。

Parameters:

board (List[str]) – 当前游戏棋盘，以字符串列表形式表示。
is_maximizing (bool) - 如果为True表示最大化玩家(O)，如果为False表示最小化玩家(X)。
depth (int) – 搜索树中的当前深度。(默认值: 0)
alpha (float) - 用于剪枝的alpha值。(默认: -math.inf)
beta (float) – 用于剪枝的Beta值。(默认: math.inf)

Returns:

包含以下元素的元组：

float: 最佳移动的得分（1表示O获胜，-1表示X
获胜，0表示平局）
Optional[int]: 最佳移动的索引，如果是终局则为
None

Return type:

元组[浮点数, 可选整数]

select_move(board: List[str]) → int | None[来源]#

根据对手的游戏风格选择一步棋。

Parameters:

board (List[str]) – 当前游戏棋盘，以字符串列表形式表示。

Returns:

所选移动的索引，如果没有移动则为None: 可用。

Return type:

可选[int]

class camel.environments.SingleStepEnv(dataset: StaticDataset | BaseGenerator, verifier: BaseVerifier, **kwargs)[来源]#

基类：object

一个轻量级环境，用于单步强化学习，使用LLMs作为策略。

该环境模拟了一个基于LLM的智能体与来自数据集的问题（如问答或数学问题）之间的单次交互过程，其中智能体生成一个响应并接收反馈。

Core Flow:

从（可能无限长的）数据集中采样一个问题。
LLM生成单步响应（即动作）。
响应将与真实情况进行比对验证。
奖励是基于正确性和可选的自定义逻辑计算的。

Key Features:

支持批量评估并跟踪每个样本的状态。
验证器及相关资源的异步设置和拆卸。
支持通过本地RNG进行确定性采样（可选种子）。
通过子类化实现可扩展的奖励计算。

ACCURACY_REWARD = 1#

PLACEHOLDER_OBS = Observation(question='Episode ended. This is just a placeholder.', context={}, metadata=None)#

async close() → None[来源]#

清理并关闭环境使用的所有资源。

该方法会关闭验证器，重置内部状态，并确保环境被正确关闭。

Raises:: 异常 – 如果在关闭环境时发生错误。

property metadata: Dict[str, Any]#

检索环境的元数据。

这提供了额外的参数和配置详情。

Returns:: 环境元数据的副本。
Return type:: 字典[字符串, 任意类型]

async reset(batch_size: int = 1, seed: int | None = None) → 观察 | List[观察][来源]#

重置环境并开始一个新的回合。

该方法从数据集中采样一批新的数据点，并返回相应的初始观测值。

如果提供了种子值，将初始化一个本地随机数生成器用于确定性采样。全局随机状态不会受到影响。

Parameters:

batch_size (int) - 要采样的数据点数量。 (默认值: 1)
seed (可选[int]) - 用于确定性采样的种子值。如果为None，则采样是非确定性的。(默认值: None)

Returns:

该事件的初步观察: 情节。

Return type:

Observation 或 List[Observation]

Raises:

RuntimeError - 如果在处理完所有先前状态之前调用。
ValueError - 如果批处理大小超过数据集大小。
TypeError - 如果数据集类型不受支持。

async setup() → None[来源]#

通过初始化验证器来设置环境。

该方法确保环境已准备好进行交互。它会设置必要的组件，包括验证器。

Raises:: 异常 - 如果由于内部错误导致设置失败。

async step(action: 操作 | List[操作] | str | Dict[int, str]) → Tuple[观察, float, bool, Dict[str, Any]] | List[Tuple[观察, float, bool, Dict[str, Any]]][来源]#

使用提出的解决方案在环境中执行一次交互步骤。

该方法处理智能体对当前观察结果的响应，使用验证器验证响应的正确性，计算奖励，并返回最终的状态转换结果。

环境严格遵循单步执行。一旦为某个状态提交了动作，该状态即被标记为完成，且观察结果将不再改变。

Parameters:

action (Union[Action, List[Action], str, Dict[int, str]]) –

智能体采取的动作，: 应包含对观察的响应

可以是以下形式： - 单个Action对象（批大小为1时） - Action对象列表（用于批量评估） - 原始字符串（仅当批大小为1时允许） - 将索引映射到其llm_response的字典

（用于批量评估）

Returns:

一个元组或元组列表，包含： - Observation: 表示回合结束的占位符。 - float: 响应的奖励值。 - bool: 回合是否结束

(在本例中始终为True)。

dict: 附加信息，包括提出的解决方案，
验证结果和原始数据点。

Return type:

联合类型[元组[Observation, 浮点数, 布尔值, 字典[字符串, 任意类型]], 列表[…]]

Raises:

RuntimeError – 如果环境尚未设置，或者未调用reset()。
ValueError - 如果检测到无效的操作格式、重复索引或越界索引。

class camel.environments.StepResult(*, observation: ~camel.environments.models.Observation, reward: float, rewards_dict: ~typing.Dict[str, float] = <factory>, done: bool, info: ~typing.Dict[str, ~typing.Any] = <factory>)[来源]#

基类：BaseModel

环境步骤的执行结果。

observation#

下一个观察结果。

Type:: camel.environments.models.Observation

reward#

不同方面的奖励分数字典。

Type:: 浮点数

done#

该剧集是否已完结。

Type:: 布尔值

info#

关于该步骤的附加信息。

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

as_tuple() → Tuple[观察, float, bool, Dict[str, Any]][来源]#: 以元组形式返回模型的所有字段，按声明顺序排列

done: bool#

info: Dict[str, Any]#

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

observation: 观察#

reward: float#

rewards_dict: Dict[str, float]#

class camel.environments.TicTacToeEnv(extractor: BaseExtractor | None = None, max_steps: int | None = None, play_style: Literal['optimal', 'random'] = 'optimal', **kwargs)[来源]#

基类: MultiStepEnv

一个用于强化学习的Tic Tac Toe（井字棋）环境，专为LLMs设计。

这个环境实现了一个标准的井字棋游戏，其中LLM智能体作为'X'与作为'O'的AI对手对战。对手可以采用最优策略（带alpha-beta剪枝的minimax算法）或随机策略。

WIN_COMBINATIONS: ClassVar = [(0, 1, 2), (3, 4, 5), (6, 7, 8), (0, 3, 6), (1, 4, 7), (2, 5, 8), (0, 4, 8), (2, 4, 6)]#

static available_moves(board: List[str]) → List[int][来源]#

获取棋盘上所有可行的走法。

Parameters:: board (List[str]) – 当前游戏棋盘，以字符串列表形式表示。
Returns:: 表示棋盘上空单元格的索引列表。
Return type:: 整数列表

static check_winner(board: List[str]) → Literal['X', 'O', 'draw'] | None[来源]#

检查棋盘上是否有赢家或平局。

Parameters:

board (List[str]) – 当前游戏棋盘，以字符串列表形式表示。

Returns:

如果X获胜则显示"X"，如果O获胜则显示"O": 如果比赛平局则显示"draw"，如果比赛仍在进行中则显示None。

Return type:

可选[字面量["X", "O", "平局"]]

async compute_reward() → Tuple[float, Dict[str, float]][来源]#

计算当前状态的奖励值。

Returns:

包含总奖励的元组: 以及奖励组件的字典： - 胜利得1.0分 - 失败或非法移动得0.0分 - 平局得0.5分 - 对于进行中的游戏，返回对局面的评估

Return type:

元组[浮点数, 字典[字符串, 浮点数]]

static evaluate_position_for_x(board: List[str], is_x_turn: bool, depth: int = 0, max_depth: int = 10) → float[来源]#

从X的角度评估当前棋盘局势。

使用极小极大算法来确定该位置的价值。

Parameters:

board (List[str]) – 当前游戏棋盘，以字符串列表形式表示。
is_x_turn (bool) - 如果轮到X走棋则为True，否则为False。

Returns:

表示位置评估的浮点数值：

如果X处于获胜位置则为1.0
如果O处于获胜位置则为0.0
平局时为0.5
对于未结束的位置，返回在
完美对弈情况下的预期结果

Return type:

浮点数

render_board(board: List[str]) → str[来源]#

将棋盘渲染为字符串以供显示。

Parameters:: board (List[str]) – 当前游戏棋盘，以字符串列表形式表示。
Returns:: 棋盘的格式化字符串表示。
Return type:: 字符串