文档
组件
Agentic
执行环境（Envs）

执行环境（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)  # 额外元数据

两种使用模式

  1. 交互模式（多轮 rollout）—— 逐步执行：
env = MyEnv()
env.reset(trajectory)
result = env.step('search', {'query': 'Python'})
# ... 重复直到 result.done
  1. 批量评估模式 —— 评估已完成的轨迹：
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_urlstr运行中的 OpenEnv 服务器 URL。
env_clsstr 或 class类型化客户端的点分导入路径或类。None 使用 GenericEnvClient。
env_kwargsDict传递给客户端构造函数的额外参数。
tool_schemaList[ToolInfo]通过 tools() 暴露的工具定义。
action_mapperCallable自定义函数，将 (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()
docs