[rollout, trainer] feat: extend agent loop for custom implementations

verl/experimental/agent_loop/tool_agent_loop.py

@@ -30,7 +30,7 @@
     register,
 )
 from verl.experimental.agent_loop.tool_parser import FunctionCall, ToolParser
-from verl.experimental.agent_loop.utils import add_generation_prompt_for_gpt_oss, format_gpt_oss_tool_response_manually
+from verl.experimental.agent_loop.utils import build_gpt_oss_tool_response_text
 from verl.interactions.base import BaseInteraction
 from verl.interactions.utils.interaction_registry import initialize_interactions_from_config
 from verl.tools.schemas import ToolResponse
@@ -360,14 +360,7 @@ async def _handle_processing_tools_state(self, agent_data: AgentData) -> AgentSt
         else:
             if self.tool_parser_name == "gpt-oss":
                 logger.info("manually format tool responses for gpt-oss")
-                # Format tool responses manually
-                tool_response_texts = []
-                for i, tool_msg in enumerate(add_messages):
-                    actual_tool_name = tool_call_names[i]
-                    formatted = format_gpt_oss_tool_response_manually(tool_msg["content"], actual_tool_name)
-                    tool_response_texts.append(formatted)
-
-                tool_response_text = add_generation_prompt_for_gpt_oss("".join(tool_response_texts))
+                tool_response_text = build_gpt_oss_tool_response_text(add_messages, tool_call_names)
                 response_ids = await self.loop.run_in_executor(
                     None, lambda: self.tokenizer.encode(tool_response_text, add_special_tokens=False)
                 )

verl/experimental/agent_loop/utils.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import os
+from typing import Any
 
 
 def resolve_config_path(config_path: str) -> str:
@@ -95,3 +96,13 @@ def add_generation_prompt_for_gpt_oss(message_content: str) -> str:
         Message content string with generation prompt
     """
     return message_content + "<|start|>assistant"
+
+
+def build_gpt_oss_tool_response_text(messages: list[dict[str, Any]], tool_call_names: list[str]) -> str:
+    """Build gpt-oss tool response text (manual formatting + generation prompt)."""
+    tool_response_texts: list[str] = []
+    for i, tool_msg in enumerate(messages):
+        actual_tool_name = tool_call_names[i]
+        formatted = format_gpt_oss_tool_response_manually(tool_msg["content"], actual_tool_name)
+        tool_response_texts.append(formatted)
+    return add_generation_prompt_for_gpt_oss("".join(tool_response_texts))

verl/trainer/ppo/ray_trainer.py

@@ -55,6 +55,7 @@
 from verl.utils.checkpoint.checkpoint_manager import find_latest_ckpt_path, should_save_ckpt_esi
 from verl.utils.config import omega_conf_to_dataclass
 from verl.utils.debug import marked_timer
+from verl.utils.import_utils import load_class_from_fqn
 from verl.utils.metric import reduce_metrics
 from verl.utils.py_functional import rename_dict
 from verl.utils.rollout_skip import RolloutSkip
@@ -847,7 +848,12 @@ def init_workers(self):
         # create async rollout manager and request scheduler
         self.async_rollout_mode = False
         if self.config.actor_rollout_ref.rollout.mode == "async":
-            from verl.experimental.agent_loop import AgentLoopManager
+            # Support custom AgentLoopManager via config
+            manager_class_fqn = self.config.actor_rollout_ref.rollout.get("agent", {}).get("agent_loop_manager_class")
+            if manager_class_fqn:
+                AgentLoopManager = load_class_from_fqn(manager_class_fqn, "AgentLoopManager")
+            else:
+                from verl.experimental.agent_loop import AgentLoopManager
 
             self.async_rollout_mode = True
             if self.config.reward_model.enable and self.config.reward_model.enable_resource_pool:

verl/utils/import_utils.py

@@ -197,6 +197,39 @@ def load_extern_object(module_path: str, object_name: str) -> object:
     return getattr(module, object_name)
 
 
+def load_class_from_fqn(fqn: str, description: str = "class") -> type:
+    """Load a class from its fully qualified name.
+
+    Args:
+        fqn: Fully qualified class name (e.g., 'mypackage.module.ClassName').
+        description: Description for error messages (e.g., 'AgentLoopManager').
+
+    Returns:
+        The loaded class.
+
+    Raises:
+        ValueError: If fqn format is invalid (missing dot separator).
+        ImportError: If the module cannot be imported.
+        AttributeError: If the class is not found in the module.
+
+    Example:
+        >>> cls = load_class_from_fqn("verl.experimental.agent_loop.AgentLoopManager")
+        >>> instance = cls(config=config, ...)
+    """
+    if "." not in fqn:
+        raise ValueError(
+            f"Invalid {description} '{fqn}'. Expected fully qualified class name (e.g., 'mypackage.module.ClassName')."
+        )
+    try:
+        module_path, class_name = fqn.rsplit(".", 1)
+        module = importlib.import_module(module_path)
+        return getattr(module, class_name)
+    except ImportError as e:
+        raise ImportError(f"Failed to import module '{module_path}' for {description}: {e}") from e
+    except AttributeError as e:
+        raise AttributeError(f"Class '{class_name}' not found in module '{module_path}': {e}") from e
+
+
 @deprecated(replacement="load_module(file_path); getattr(module, type_name)")
 def load_extern_type(file_path: str, type_name: str) -> type:
     """DEPRECATED. Directly use `load_extern_object` instead."""

verl/workers/config/rollout.py

@@ -71,6 +71,9 @@ class AgentLoopConfig(BaseConfig):
     default_agent_loop: str = "single_turn_agent"
     agent_loop_config_path: Optional[str] = None
     custom_async_server: CustomAsyncServerConfig = field(default_factory=CustomAsyncServerConfig)
+    # Fully qualified class name for custom AgentLoopManager (e.g., "mypackage.module.MyManager").
+    # Security: This class will be dynamically imported via importlib. Only use trusted class paths.
+    agent_loop_manager_class: Optional[str] = None
 
 
 @dataclass
@@ -179,6 +182,9 @@ class RolloutConfig(BaseConfig):
     # Use Prometheus to collect and monitor rollout statistics
     prometheus: PrometheusConfig = field(default_factory=PrometheusConfig)
 
+    # Extension point for custom configurations
+    custom: Optional[dict] = None
+
     update_weights_bucket_megabytes: int = 512
 
     skip_rollout: bool = False