[RFC] Agent Abstractions and Trajectory Gateway for VERL

[zackcxb]

[RFC] Agent Abstractions and Trajectory Gateway for VERL

Summary

This RFC proposes two new abstractions for VERL's agent-based reinforcement learning pipeline:

1. AgentFramework — an abstract base class for agent lifecycle management and reward computation, replacing the current tight coupling between AgentLoopManager and specific agent implementations.
2. AgentGateway — a Gateway subsystem owned by the serving layer (LLMServerManager) that intercepts agent LLM calls via the OpenAI Chat Completions API, performs canonical tokenization, and assembles token-level trajectory data with strict token-truth guarantees.

Together, they enable any OpenAI-compatible agent system to be integrated into VERL's training loop without modifications to agent code, while producing continuous multi-turn trajectories with loss masks directly consumable by VERL's training engine.

Motivation

VERL's current agent integration (AgentLoopManager + AgentLoopBase) tightly couples three concerns: LLM infrastructure management, agent lifecycle, and trajectory collection. This creates friction when integrating new agent types:

• Each new agent framework requires dedicated adapter code inside the agent loop.
• Trajectory collection logic is embedded in the agent loop itself, making it non-reusable.
• Only coroutine-based agents are natively supported; subprocess and remote agents require ad-hoc integration (e.g., SWE-Agent's custom ModelProxy).

Community contributions such as AWS AgentCore (PR #4216) and Aliyun Remote Agent (Issue #5737) further demonstrate the need for a pluggable agent abstraction that cleanly separates these concerns.

This RFC addresses these issues by:

1. Defining AgentFramework as a thin standard interface for agent-based rollout. The common contract is generate_sequences(prompts: DataProto) -> DataProto; internal execution structure, reward computation flow, and batching strategy remain implementation-specific.
2. Extracting trajectory collection into AgentGateway, a serving-side subsystem that works with any Framework implementation. The Gateway handles tokenization, prefix consistency, and trajectory assembly — concerns that are orthogonal to how agents are launched or managed.
3. Extracting infrastructure management (LLM server initialization, load balancing) out of the framework layer, so that AgentFramework remains focused on trainer-facing generation semantics rather than serving ownership.

Design Overview

Architecture

VERL Training Loop
│ 
│
├── AgentFramework
│     Agent lifecycle management
│     Reward computation
│     Batch orchestration + DataProto assembly
│
└── Serving Runtime
      Owns: LLMServerManager, load balancer, Gateway subsystem
      ├── GatewayManager (internal session-routing component)
      ├── Gateway Actor 1 (Ray actor, FastAPI)
      ├── Gateway Actor 2 (Ray actor, FastAPI)
      └── Gateway Actor N (Ray actor, FastAPI)

AgentFramework and AgentGateway are independent — the Gateway does not know which Framework implementation is using it, and the Framework does not know the Gateway's internal trajectory assembly logic. They interact only through a well-defined session API.

To avoid single-point bottlenecks, multiple Gateway instances run as Ray actors, each hosting a FastAPI HTTP server. An AgentGatewayManager routes session creation requests across Gateway actors (e.g., round-robin or least-loaded). Once a session is created on a specific Gateway actor, all subsequent requests for that session are pinned to that actor. This follows the same pattern as VERL's existing GlobalRequestLoadBalancer for LLM server replicas.

Data Flow

A single session proceeds as follows:

1. Framework creates a session on the AgentGatewayManager, which selects a Gateway actor and returns a session-specific base_url.
2. Framework starts the agent (subprocess, coroutine, or remote call), injecting the base_url as the agent's LLM endpoint.
3. Agent makes standard OpenAI Chat Completion requests to the assigned Gateway actor. On each request, the Gateway tokenizes, checks prefix consistency, routes to the inference backend, records the token-level interaction, and returns a standard OpenAI response. The agent is unaware of the interception.
4. Agent completes (process exits, coroutine returns, or calls the optional /complete endpoint).
5. Framework finalizes the session via the AgentGatewayManager, receiving the assembled trajectories.
6. Framework computes trajectory-aligned rewards and packages the resulting samples into DataProto.

AgentFramework

Interface

class AgentFramework(ABC):
    @abstractmethod
    async def generate_sequences(self, prompts: DataProto) -> DataProto:
        """Process a trainer batch and return a training-ready DataProto."""
        ...

AgentFramework is intentionally thin. Users are expected to implement agent lifecycle management, reward computation, and batch orchestration through the generate_sequences interface. Common patterns may be factored into subclasses and helpers where useful.

This keeps the interface compatible with heterogeneous execution models:

• VERL-native agent loops
• subprocess-based OpenAI-compatible agents
• remote services and cloud-hosted agent frameworks

Reward Computation

Reward assignment remains a shared concern, but not a required abstract method on AgentFramework.

Typical patterns include:

• Framework-collected. When the framework can directly access agent output (subprocess stdout, coroutine return value), it parses the relevant information locally.
• Agent-uploaded. When the agent runs remotely and the framework has no direct access to its output, the agent uploads reward_info via the Gateway's optional /complete endpoint, and the framework reads it back during session finalization.
• Helper-normalized. Framework implementations may reuse shared helpers to normalize one session-level reward into per-trajectory rewards, validate trajectory/reward alignment, and assemble DataProto.

Reference Implementations

OpenAICompatibleAgentFramework is the preferred first implementation target. It launches or contacts an OpenAI-compatible agent, injects a session base_url, and relies on Gateway-backed /v1/chat/completions traffic as the trajectory truth source.

CliAgentFramework is a straightforward variant that launches external agent programs as subprocesses, injecting the Gateway session URL via the OPENAI_BASE_URL environment variable. Any agent that uses the OpenAI Chat Completions API works without code changes. Completion is detected via process exit.

AgentLoopManager remains an important follow-up migration target, but it is not part of the first implementation milestone. Supporting it cleanly will likely require a token-request ingress in addition to the chat-completions path described in this RFC.

Custom implementations can support other execution models such as remote services or cloud-hosted frameworks by using the same session API. For remote agents without an external notification channel, the Gateway provides wait_for_completion() to block until the agent signals completion via /complete.

Migration from AgentLoopManager

The current AgentLoopManager remains on the legacy path in the first implementation stage.

The staged plan is:

• First, land a Gateway-backed chat-completions path for OpenAI-compatible / remote-style agents.
• Then, add token-request ingress as a dedicated extension point.
• Finally, migrate AgentLoopManager and related VERL-native agent loops onto the new serving/Gateway model without introducing double trajectory bookkeeping in production.

AgentGateway

Overview

Each AgentGateway instance is a Ray actor running a FastAPI HTTP server that exposes the OpenAI Chat Completions API. It manages multiple concurrent sessions, each maintaining independent trajectory state. The Gateway is the single canonical tokenization authority for the chat-completions path — all messages -> token_ids conversions happen here, using the inference backend's tokenizer and chat template.

In the first implementation stage, /v1/chat/completions is the primary ingress. A token-request ingress for legacy VERL-native paths is reserved as a follow-up extension.

Multiple Gateway actors are managed by an AgentGatewayManager, which handles session routing and provides a unified interface to the Framework.

Gateway Manager and Scaling

Each Gateway actor manages its own sessions independently. The manager's only responsibility is routing — selecting which actor handles a new session and forwarding subsequent calls to the correct actor.

class AgentGatewayManager:
    """Manages multiple Gateway actors with session routing."""

    def __init__(self, gateways: list[AgentGateway]):
        self.gateways = gateways
        self._session_to_gateway: dict[str, AgentGateway] = {}

    async def create_session(self, session_id: str) -> GatewaySession:
        """Select a Gateway actor (e.g., round-robin) and create a session."""
        gateway = self._select_gateway()
        session = await gateway.create_session.remote(session_id)
        self._session_to_gateway[session_id] = gateway
        return session

    async def finalize_session(self, session_id: str) -> list[Trajectory]:
        """Route to the correct Gateway actor and finalize."""
        gateway = self._session_to_gateway.pop(session_id)
        return await gateway.finalize_session.remote(session_id)

    async def abort_session(self, session_id: str) -> None:
        gateway = self._session_to_gateway.pop(session_id, None)
        if gateway:
            await gateway.abort_session.remote(session_id)

    async def wait_for_completion(self, session_id: str, timeout: float) -> None:
        gateway = self._session_to_gateway[session_id]
        await gateway.wait_for_completion.remote(session_id, timeout)

Session Management

The Gateway provides a session API for Framework to manage session lifecycles:

@ray.remote
class AgentGateway:

    def __init__(
        self,
        tokenizer: AutoTokenizer,
        chat_template: str,
        backend: InferenceBackend,
        config: GatewayConfig,
    ): ...

    async def create_session(self, session_id: str) -> GatewaySession:
        """Create a trajectory session."""
        ...

    async def finalize_session(self, session_id: str) -> list[Trajectory]:
        """Assemble and return trajectories, clean up session state.
        Returns one trajectory per prefix-consistent segment."""
        ...

    async def abort_session(self, session_id: str) -> None:
        """Discard session state."""
        ...

    async def wait_for_completion(self, session_id: str, timeout: float) -> None:
        """Block until agent calls /complete. For remote agents only."""
        ...

create_session returns a GatewaySession containing a session-specific base_url (e.g., http://{gateway_host}:{port}/sessions/{session_id}/v1). The agent uses this URL for all LLM calls, and the Gateway routes requests to the correct session by URL path. This approach requires no special headers or client modifications — the agent simply uses a different base URL.

HTTP Endpoints

The Gateway exposes two endpoints per session:

POST /sessions/{id}/v1/chat/completions — Standard OpenAI Chat Completions. The agent calls this as its normal LLM endpoint. The Gateway intercepts the request, performs tokenization, routes to the inference backend, records the interaction, and returns a standard response. This is the mandatory first-stage endpoint.

POST /sessions/{id}/complete — Optional. Allows the agent to explicitly signal session completion and optionally upload reward-related information. This is useful for remote agents that have no other completion notification channel, or for VERL-native agent loops that want to pass structured results. Agents that do not call this endpoint are unaffected — Framework detects completion through other means (process exit, coroutine return, etc.).

Request Handling

On each Chat Completion request, the Gateway performs:

1. Message-level prefix check. Compare the incoming normalized messages with the session's recorded message history. If the prefix matches, only the new incremental messages are tokenized and appended to the accumulated token sequence. If the prefix does not match, the current trajectory is finalized and a new trajectory begins with full tokenization.
2. Inference routing. Send prompt_ids to the inference backend via its token-level generation API (e.g. AsyncLLMServerManager). Receive response_ids and logprobs.
3. Interaction recording. Record the turn's prompt_ids, response_ids, and logprobs. Update the session's accumulated token sequence and message history.
4. Response reconstruction. Detokenize the response and construct a standard OpenAI Chat Completion response for the agent. When tool calling is enabled, structured tool_calls fields are reconstructed from the raw token output.

Trajectory Output

After a session completes, finalize_session assembles the recorded interactions into a list[Trajectory]. Each trajectory is a prefix-consistent, continuous token sequence that constitutes an independent training sample:

@dataclass
class Trajectory:
    uid: str                         # Each prompt has a unique uuid from dataset
    session_id: int                  # Each group sampling has a session_id: [0, n)
    trajectory_id: int               # Each sampling outputs m trajectories: [0, m)
    reward_info: dict

    prompt_ids: list[int]
    response_ids: list[int]
    response_logprobs: list[float]
    loss_mask: list[int]             # 1 for response tokens, 0 for prompt
    ...

A session produces multiple trajectories when the Gateway detects a message prefix mismatch mid-session, possibly due to context compression, skill switching, or other agent-side context rewrites. The Gateway does not need to understand why the context changed; it only enforces consistency within each trajectory. In the final DataProto output, each trajectory becomes one row.

Extensions

Multimodal Support

The Gateway architecture supports multimodal inputs via an optional preprocessor. When present, the Gateway applies the multimodal preprocessor during tokenization and stores processor outputs alongside token sequences. Specific processor adapters will be added as model support grows.

Tool Call Reconstruction

For models that produce tool calls via special tokens, the Gateway uses a configurable tool parser to reconstruct structured tool_calls fields in the OpenAI response returned to the agent. The raw token sequence is always preserved as-is for training.

Token-Request Ingress

A token-request ingress is reserved for follow-up work so that legacy VERL-native paths such as AgentLoopManager can migrate onto the same Gateway/session model without keeping legacy trajectory bookkeeping as a production truth source. This extension is explicitly out of scope for the first implementation stage.

Prefix-Sharing Storage

In scenarios with repeated sampling or partial context overlap, multiple trajectories may share common prefixes. Tree-structured storage could reduce memory and disk usage, but training-side benefits depend on algorithm-level support (e.g., DTA). This is deferred to future work pending further analysis.

Update on May 11

#6299 is the most updated draft PR, superseding PR #5931.

What's done:

• The core implementation for AgentGateway, including the gateway actor and the gateway serving runtime.
• A high-level example implementation of an OpenAI-request-compatible framework
• adaptation to the new main_ppo_sync.py entrance based on Transfer Queue.
• Multi-modal and tool parsing support
• Many AI-generated unit tests and a few smoke tests that may need to be trimmed down.

WIP:

• A deepeyes_with_gateway recipe
• A SWE-agent recipe
• A CLIAgentFramework example implementation
• Integrate Gateway/Framework configs into the current VERL config system
• CI hygiene

Future directions:

• Turn-wise/completion-wise trajectory collection
• Multi-agent support
• Default deployment strategy for gateway actors

[wuxibin89]

External agent frameworks integration request

• [RFC] A Fully decoupled and auto-scaled rollout engine using AWS Bedrock AgentCore Runtime #4216
• Feature Request: RemoteAgentLoop - Support for External Distributed Agent Integration #5737
• [worker, vllm] feat: expose nemo gym token id patch #5833

[SendoRay]

Hi @zackcxb . I'd like to volunteer to implement this. After studying the current codebase (especially AgentLoopManager, AgentLoopWorker, and the existing agent_loop_manager), I've put together a detailed implementation plan.

architechture

ray_trainer.py
  |
  |  self.async_rollout_manager = await AgentFramework.create(...)
  |  self.checkpoint_manager = CheckpointEngineManager(replicas=self.async_rollout_manager.rollout_replicas)
  |  self.async_rollout_manager.generate_sequences(DataProto) -> DataProto
  |
  AgentFramework (abstract)
  |  职责: agent 生命周期、reward 计算、trajectory -> DataProto 组装
  |  接口: generate_sequences(DataProto) -> DataProto
  |         rollout_replicas (property, 代理到内部 GatewayManager)
  |
  |  framework 内部通过 create_session / finalize_session 与 GatewayManager 交互
  |
  AgentGatewayManager
  |  职责: 拥有 rollout replicas、GlobalRequestLoadBalancer、多个 Gateway actors
  |  接口: create_session(session_id) -> GatewaySession
  |         finalize_session(session_id) -> list[Trajectory]
  |         rollout_replicas (property, 暴露给 CheckpointEngineManager)
  |         wake_up() / sleep() / clear_kv_cache() / start_profile() / stop_profile()
  |
  AgentGateway (Ray actor, FastAPI)
  |  职责: tokenize、prefix consistency、trajectory recording、HTTP proxy
  |  转发请求到 rollout server (通过 load balancer)

核心设计原则：

• trainer 直接持有 AgentFramework 实例，通过 generate_sequences 交互
• AgentFramework 通过 AgentGatewayManager 的 session API 与 Gateway 交互
• AgentGatewayManager 拥有所有 serving runtime（rollout replicas + Gateway actors）
• AgentFramework 本身就是 trainer 的直接交互对象

Trainer 集成方式

ray_trainer.py 中的修改：当 agent.framework.implementation 配置存在时，直接创建 AgentFramework 实例（而非 AgentLoopManager）：

# 新增分支（在 agent_loop_manager_class 之前检查）
framework_fqn = config.actor_rollout_ref.rollout.agent.framework.implementation
if framework_fqn:
    AgentFramework = load_class_from_fqn(framework_fqn)
    self.async_rollout_manager = await AgentFramework.create(config, worker_group, ...)
elif manager_class_fqn:
    ...

---

分阶段实现计划

Phase 1: 基础抽象层

目标： 定义 AgentFramework ABC、AgentGatewayManager、AgentGateway 的接口。

新建文件：

verl/experimental/agent_framework/
    __init__.py
    base.py                       # AgentFramework ABC
    gateway.py                    # AgentGateway (Ray actor, FastAPI)
    gateway_manager.py            # AgentGatewayManager
    gateway_tokenizer.py          # Tokenization + prefix consistency
    session_manager.py            # Session state management
    config.py                     # 配置 dataclass

修改文件：

• verl/workers/config/rollout.py — 添加 AgentFrameworkConfig 到 AgentLoopConfig
• verl/trainer/ppo/ray_trainer.py — 添加 agentframework 分支

AgentFramework 接口：

class AgentFramework(ABC):
    @classmethod
    @abstractmethod
    async def create(cls, config, worker_group=None, rollout_resource_pool=None, ...):
        """创建 framework 实例，内部创建 GatewayManager。"""
        ...

    @abstractmethod
    async def generate_sequences(self, prompts: DataProto) -> DataProto:
        """处理 trainer batch，返回训练就绪的 DataProto。"""
        ...

    @abstractmethod
    async def wake_up(self): ...

    @abstractmethod
    async def sleep(self): ...

    @abstractmethod
    async def clear_kv_cache(self): ...

    @abstractmethod
    async def start_profile(self, **kwargs): ...

    @abstractmethod
    async def stop_profile(self): ...

    @property
    @abstractmethod
    def rollout_replicas(self) -> list:
        """暴露 rollout replicas 给 CheckpointEngineManager。"""
        ...

AgentGatewayManager 接口：

class AgentGatewayManager:
    """拥有 rollout replicas、load balancer、Gateway actors。
    提供 session API 给 AgentFramework 使用。"""

    async def initialize(self, config, worker_group=None):
        """初始化 rollout replicas + load balancer + Gateway actors。"""
        ...

    async def create_session(self, session_id: str) -> GatewaySession:
        """选择一个 Gateway actor，创建 session，返回 session-specific base_url。"""
        ...

    async def finalize_session(self, session_id: str) -> list[TrajectorySegment]:
        """从 Gateway actor 获取 assembled trajectories。"""
        ...

    async def abort_session(self, session_id: str): ...

    @property
    def rollout_replicas(self) -> list: ...

    async def wake_up(self): ...
    async def sleep(self): ...
    async def clear_kv_cache(self): ...
    async def start_profile(self, **kwargs): ...
    async def stop_profile(self): ...

AgentGateway 接口：

@ray.remote(num_cpus=2)
class AgentGateway:
    """Ray actor，运行 FastAPI server。
    单个 Gateway 管理多个 concurrent sessions。"""

    async def create_session(self, session_id: str) -> GatewaySession: ...
    async def finalize_session(self, session_id: str) -> list[TrajectorySegment]: ...
    async def abort_session(self, session_id: str): ...

    # Ray remote call 模式
    async def generate(self, request_id, prompt_ids, sampling_params, **kwargs) -> TokenOutput: ...

    # HTTP endpoint (Phase 3)
    # POST /sessions/{id}/v1/chat/completions
    # POST /sessions/{id}/complete

---

Phase 2: Session 管理 + Tokenization + Direct Generate

目标： 实现完整的 session 管理、tokenization、prefix consistency、直接 generate 路径。

核心实现：

SessionManager:

class SessionState:
    session_id: str
    prefix_token_ids: list[int]
    message_history: list[dict]
    segments: list[TrajectorySegment]
    reward_info: dict | None

class SessionManager:
    def create_session(self, session_id) -> SessionState: ...
    def record_interaction(self, session_id, prompt_ids, response_ids, response_mask, logprobs): ...
    def check_prefix(self, session_id, new_messages) -> (bool, int): ...
    def finalize(self, session_id) -> list[TrajectorySegment]: ...

GatewayTokenizer:

class GatewayTokenizer:
    def tokenize_messages(self, messages, tools=None, images=None) -> list[int]: ...
    def check_prefix_consistency(self, stored, new) -> (bool, int): ...
    def detokenize_response(self, response_ids) -> str: ...

---

Phase 3: OpenAI-Compatible 完整实现

目标： 实现 OpenAICompatibleAgentFramework、HTTP endpoint、tool call reconstruction、DataProto 组装。

新建文件：

verl/experimental/agent_framework/
    openai_compatible.py          # OpenAICompatibleAgentFramework
    cli_agent.py                  # CliAgentFramework
    gateway_tool_reconstructor.py # Tool call reconstruction
    data_proto_assembler.py       # Trajectory -> DataProto 组装

OpenAICompatibleAgentFramework 流程：

1. generate_sequences(DataProto) 被 trainer 调用
2. 对 batch 中每个 sample：
   a. 调用 gateway_manager.create_session(session_id) 获取 Gateway session
   b. 将 raw_prompt messages 发送到 Gateway
   c. 多轮循环直到 agent 终止
   d. 调用 gateway_manager.finalize_session(session_id) 获取 trajectory
3. 计算 reward
4. 将 trajectory 组装为 DataProto（与 AgentLoopWorker._postprocess() 输出格式一致）

Gateway HTTP endpoints:

• POST /sessions/{id}/v1/chat/completions — OpenAI Chat Completions 代理
• POST /sessions/{id}/complete — agent 完成信号

Tool call reconstruction: 复用现有 ToolParser，新增 reconstruct() 方法将 FunctionCall 转为 OpenAI tool_calls 格式。

---

Phase 4: CliAgentFramework + 测试 + 配置示例

CliAgentFramework：

1. 创建 Gateway session
2. 启动子进程，注入 OPENAI_BASE_URL=<session_base_url>
3. 等待进程退出
4. finalize_session() 获取 trajectory
5. 组装为 DataProto

测试：

• Unit tests: GatewayTokenizer, SessionManager, prefix consistency, tool call reconstruction
• Integration test: 对比 OpenAICompatibleAgentFramework 与 AgentLoopManager 输出的 DataProto 格式
• E2E test: CliAgentFramework + 简单 OpenAI-compatible 脚本
• Regression: 现有 agent loop 测试全部通过

---

如果您有什么建议或者改进点，请随时联系我。

[zackcxb]

@SendoRay Hi there, thank you so much for volunteering to implement this RFC!
We’ve already prepared a draft PR and plan to open it by tomorrow. Our implementation is largely consistent with your suggestions, with only minor differences in module ownership.
We’d really welcome your comments and contributions to the PR once it’s published!

[hhaAndroid]

在轨迹还没有完成情况下，AgentGateway 需要暂存并发请求情况下所有中间数据？为了避免中心化压力，是不是默认设置这个 ray actor 就是至少 num gpu worker 个，并且强制分散编排，否则这个压力也是很大的。是否将 AgentGateway 做成无状态的会更好？

[zackcxb]

在轨迹还没有完成情况下，AgentGateway 需要暂存并发请求情况下所有中间数据？为了避免中心化压力，是不是默认设置这个 ray actor 就是至少 num gpu worker 个，并且强制分散编排，否则这个压力也是很大的。是否将 AgentGateway 做成无状态的会更好？

是的，GatewayActor目前设计是要存储live session的中间数据，因为还要存储历史轨迹数据做前缀校验，所以不适合做成无状态的。当前我们GatewayActor是设计成可以起多个来分摊流量压力，不过还没有提供默认配置策略，需要用户自己提供数量，后续应该会补充一个。
由于GatewayActor是CPU侧的，直接拿num gpu worker做默认值不一定合适。我的直觉是默认最少每个节点上放一个，后续会跑一些端到端测试看下per actor承受的压力，多模态场景和纯文本场景差别可能会比较大。