执行环境（Envs）

Envs 模块提供了用于 Agentic 训练的 RL 执行环境抽象。环境可以在多轮 rollout 中交互式参与，也可以批量评估已完成的轨迹。

Env 基类

from twinkle_agentic.envs.base import Env, StepResult

class Env(ABC):

    def reset(self, trajectory=None) -> StepResult:
        """重置环境，开始新一轮。"""

    @abstractmethod
    def step(self, tool_name: str, arguments: dict) -> StepResult:
        """执行单个动作，返回观测 + 奖励 + 完成标志。"""

    def tools(self) -> List[ToolInfo]:
        """返回此环境中可用的工具定义。"""

    def evaluate(self, trajectories, **kwargs) -> List[float]:
        """批量评估已完成的轨迹，返回奖励列表。"""

    def close(self) -> None:
        """释放资源。"""

StepResult

@dataclass
class StepResult:
    observation: str = ''    # 动作执行后的环境观测
    reward: float = 0.0      # 此步骤的标量奖励
    done: bool = False        # 是否终止
    info: Dict[str, Any] = field(default_factory=dict)  # 额外元数据

两种使用模式

交互模式（多轮 rollout）—— 逐步执行：

env = MyEnv()
env.reset(trajectory)
result = env.step('search', {'query': 'Python'})
# ... 重复直到 result.done

批量评估模式 —— 评估已完成的轨迹：

rewards = env.evaluate(completed_trajectories)

EnvTool

EnvTool 将 Env 包装为 Tool，连接环境与 ToolManager 和 MultiTurnRollout。

from twinkle_agentic.envs.env_tool import EnvTool
from twinkle_agentic.tools.tool_manager import ToolManager

env = MyEnv()

# 为环境中定义的每个工具创建一个 EnvTool
env_tools = EnvTool.from_env(env)

# 注册到 ToolManager
manager = ToolManager(env_tools)

核心特性

特性	说明
`from_env(env)`	工厂方法：为 `env.tools()` 中的每个工具创建一个 `EnvTool`。
`last_result`	存储最近一次 `StepResult` 供调用方检查。
`done`	属性：最后一步是否终止了回合。
`episode_reward`	属性：来自 `info['episode_reward']` 的累计奖励。

手动构造

env_tool = EnvTool(
    env=my_env,
    tool_name='execute_code',
    description='在沙箱中执行 Python 代码。',
    parameters={
        'type': 'object',
        'properties': {
            'code': {'type': 'string', 'description': '要执行的 Python 代码。'},
        },
        'required': ['code'],
    },
)

OpenEnv

OpenEnv 将基于 WebSocket 的 OpenEnv 环境服务器适配为同步的 Twinkle Env。

from twinkle_agentic.envs.openenv import OpenEnv

env = OpenEnv(
    base_url='http://localhost:8000',
    env_cls='coding_env.CodingEnv',      # 可选的类型化客户端
    env_kwargs={'message_timeout_s': 30},
    tool_schema=[...],                    # 可选的工具定义
)

参数

参数	类型	说明
`base_url`	`str`	运行中的 OpenEnv 服务器 URL。
`env_cls`	`str` 或 class	类型化客户端的点分导入路径或类。`None` 使用 `GenericEnvClient`。
`env_kwargs`	`Dict`	传递给客户端构造函数的额外参数。
`tool_schema`	`List[ToolInfo]`	通过 `tools()` 暴露的工具定义。
`action_mapper`	`Callable`	自定义函数，将 `(tool_name, args)` 映射为发送给服务器的动作字典。

与 Rollout 集成使用

from twinkle_agentic.envs.openenv import OpenEnv
from twinkle_agentic.envs.env_tool import EnvTool
from twinkle_agentic.tools.tool_manager import ToolManager
from twinkle_agentic.rollout.api_multi_turn import APIMultiTurnRollout

# 设置环境
env = OpenEnv(base_url='http://localhost:8000', tool_schema=[...])
env.reset()

# 桥接到 ToolManager
env_tools = EnvTool.from_env(env)
manager = ToolManager(env_tools)

# 在 rollout 中使用
rollout = APIMultiTurnRollout(api=api, tool_manager=manager, max_turns=10)
results = rollout(trajectories)

实现自定义环境

from twinkle_agentic.envs.base import Env, StepResult

class CodeExecutionEnv(Env):

    def reset(self, trajectory=None):
        self._sandbox = create_sandbox()
        return StepResult(observation='沙箱已就绪。')

    def step(self, tool_name, arguments):
        code = arguments.get('code', '')
        output = self._sandbox.run(code)
        return StepResult(
            observation=output,
            reward=1.0 if 'error' not in output.lower() else 0.0,
            done=False,
        )

    def tools(self):
        return [{
            'type': 'function',
            'function': {
                'name': 'execute_code',
                'description': '运行 Python 代码。',
                'parameters': {
                    'type': 'object',
                    'properties': {
                        'code': {'type': 'string'},
                    },
                },
            },
        }]

    def close(self):
        self._sandbox.cleanup()