From e478f8a9720ca81ee649ad2e343beab0d18e48ab Mon Sep 17 00:00:00 2001 From: albertimff Date: Thu, 18 Dec 2025 20:53:11 +0800 Subject: [PATCH] [feat] Add multi-turn format checking reward managers with refactored utilities ## Files Modified: - verl/experimental/agent_loop/tool_agent_loop.py: Enhanced to record full message history for format reward computation - verl/experimental/reward/reward_loop/__init__.py: Added exports for new format check reward managers - verl/workers/reward_manager/__init__.py: Added exports for new format check reward managers ## Files Added: - verl/experimental/reward/reward_loop/format_check_dapo.py: DAPO reward manager with format checking - verl/experimental/reward/reward_loop/format_check_naive.py: Naive reward manager with format checking - verl/utils/format_reward.py: Core format reward computation utilities with helper functions - verl/workers/reward_manager/format_check_dapo.py: Worker implementation for DAPO format checking - verl/workers/reward_manager/format_check_naive.py: Worker implementation for naive format checking ## Purpose of Changes: The format reward check manager was rewritten to address multi-turn conversation scenarios where the previous implementation would concatenate all subsequent assistant/tool messages, causing single-turn format checks to become ineffective. The new solution extracts all assistant messages from the complete conversation history and performs individual format validation on each turn, ensuring proper format checking regardless of conversation length. ## Implementation Details: - **tool_agent_loop.py**: Modified to always record assistant messages and pass full message context via extra_fields - **format_reward.py**: Provides extensible format validation with regex patterns for thinking/tool/answer tags, plus refactored utility functions: - `compute_format_reward()`: Core format validation logic - `apply_format_reward_to_score()`: Helper for scalar reward scores - `apply_format_reward_to_tensor()`: Helper for tensor reward processing - **Reward Managers**: Refactored to use shared utility functions, reducing code duplication - Both DAPO and naive reward manager architectures supported - Full message history preservation enables per-turn format analysis ## Key Features: - Full message history preservation in agent loop for per-turn analysis - Configurable format checking logic that users can customize - Refactored utility functions for better code reuse - Support for both DAPO and naive reward manager architectures - No breaking API changes - fully backward compatible --- .../agent_loop/tool_agent_loop.py | 22 ++-- .../reward/reward_loop/__init__.py | 5 +- .../reward/reward_loop/format_check_dapo.py | 32 +++++ .../reward/reward_loop/format_check_naive.py | 30 +++++ verl/utils/format_reward.py | 110 ++++++++++++++++++ verl/workers/reward_manager/__init__.py | 4 + .../reward_manager/format_check_dapo.py | 47 ++++++++ .../reward_manager/format_check_naive.py | 47 ++++++++ 8 files changed, 289 insertions(+), 8 deletions(-) create mode 100644 verl/experimental/reward/reward_loop/format_check_dapo.py create mode 100644 verl/experimental/reward/reward_loop/format_check_naive.py create mode 100644 verl/utils/format_reward.py create mode 100644 verl/workers/reward_manager/format_check_dapo.py create mode 100644 verl/workers/reward_manager/format_check_naive.py diff --git a/verl/experimental/agent_loop/tool_agent_loop.py b/verl/experimental/agent_loop/tool_agent_loop.py index 4676c8513c5..000053b8891 100644 --- a/verl/experimental/agent_loop/tool_agent_loop.py +++ b/verl/experimental/agent_loop/tool_agent_loop.py @@ -166,6 +166,11 @@ async def run(self, sampling_params: dict[str, Any], **kwargs) -> AgentLoopOutpu response_ids = agent_data.prompt_ids[-len(agent_data.response_mask) :] prompt_ids = agent_data.prompt_ids[: len(agent_data.prompt_ids) - len(agent_data.response_mask)] multi_modal_data = {"image": agent_data.image_data} if agent_data.image_data is not None else {} + # for computing format reward score + # NOTE: In previous implementation, the returned `response` was a concatenation of all subsequent assistant/tool messages. + # To compute the assistant's format reward for each turn, it's better to extract assistant message info directly from `messages`. + # Therefore, we record full message information here for post-analysis. + messages = agent_data.messages output = AgentLoopOutput( prompt_ids=prompt_ids, response_ids=response_ids[: self.response_length], @@ -176,7 +181,7 @@ async def run(self, sampling_params: dict[str, Any], **kwargs) -> AgentLoopOutpu else None, num_turns=agent_data.user_turns + agent_data.assistant_turns + 1, metrics=agent_data.metrics, - extra_fields={}, + extra_fields={"messages": messages}, ) output.extra_fields.update({"turn_scores": agent_data.turn_scores, "tool_rewards": agent_data.tool_rewards}) return output @@ -242,12 +247,15 @@ async def _handle_generating_state( _, agent_data.tool_calls = await self.tool_parser.extract_tool_calls(agent_data.response_ids) # Handle interaction if needed - if self.interaction_config_file: - assistant_message = await self.loop.run_in_executor( - None, lambda: self.tokenizer.decode(agent_data.response_ids, skip_special_tokens=True) - ) - add_messages.append({"role": "assistant", "content": assistant_message}) - agent_data.messages.extend(add_messages) + # NOTE: when interaction_config_file is not enabled, the assistant messages are not recorded in the messages list. + # we should always append the full assistant messages regardless of whether interaction_config_file is enabled, + # so that assistant messages can be fully recorded for per-turn format reward computation. + # if self.interaction_config_file: + assistant_message = await self.loop.run_in_executor( + None, lambda: self.tokenizer.decode(agent_data.response_ids, skip_special_tokens=True) + ) + add_messages.append({"role": "assistant", "content": assistant_message}) + agent_data.messages.extend(add_messages) # Determine next state if agent_data.tool_calls: diff --git a/verl/experimental/reward/reward_loop/__init__.py b/verl/experimental/reward/reward_loop/__init__.py index e76197104e2..8f9d27181ed 100644 --- a/verl/experimental/reward/reward_loop/__init__.py +++ b/verl/experimental/reward/reward_loop/__init__.py @@ -15,10 +15,13 @@ from .registry import get_reward_loop_manager_cls, register # noqa: I001 from .dapo import DAPORewardLoopManager from .naive import NaiveRewardLoopManager - +from .format_check_dapo import FormatCheckDAPORewardLoopManager +from .format_check_naive import FormatCheckNaiveRewardLoopManager __all__ = [ "DAPORewardLoopManager", "NaiveRewardLoopManager", + "FormatCheckDAPORewardLoopManager", + "FormatCheckNaiveRewardLoopManager", "register", "get_reward_loop_manager_cls", ] diff --git a/verl/experimental/reward/reward_loop/format_check_dapo.py b/verl/experimental/reward/reward_loop/format_check_dapo.py new file mode 100644 index 00000000000..1c3b5f1fa55 --- /dev/null +++ b/verl/experimental/reward/reward_loop/format_check_dapo.py @@ -0,0 +1,32 @@ +# Copyright 2025 Bytedance Ltd. and/or its affiliates +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import numpy as np + +from verl import DataProto +from verl.experimental.reward.reward_loop import register +from verl.utils.format_reward import apply_format_reward_to_score +from verl.experimental.reward.reward_loop.dapo import DAPORewardLoopManager + + +@register("format_check_dapo") +class FormatCheckDAPORewardLoopManager(DAPORewardLoopManager): + """Reward loop that validates assistant message formatting and adds it to DAPO reward.""" + + def __init__(self, config, tokenizer, compute_score=None, reward_router_address=None, reward_model_tokenizer=None): + super().__init__(config, tokenizer, compute_score, reward_router_address, reward_model_tokenizer) + + async def run_single(self, data: DataProto) -> dict: + base_result = await super().run_single(data) + return apply_format_reward_to_score(data[0], base_result) diff --git a/verl/experimental/reward/reward_loop/format_check_naive.py b/verl/experimental/reward/reward_loop/format_check_naive.py new file mode 100644 index 00000000000..33dd60494a9 --- /dev/null +++ b/verl/experimental/reward/reward_loop/format_check_naive.py @@ -0,0 +1,30 @@ +# Copyright 2025 Bytedance Ltd. and/or its affiliates +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from verl import DataProto +from verl.experimental.reward.reward_loop import register +from verl.utils.format_reward import apply_format_reward_to_score +from verl.experimental.reward.reward_loop.naive import NaiveRewardLoopManager + + +@register("format_check_naive") +class FormatCheckNaiveRewardLoopManager(NaiveRewardLoopManager): + """Naive reward loop with additional format reward.""" + + def __init__(self, config, tokenizer, compute_score=None, reward_router_address=None, reward_model_tokenizer=None): + super().__init__(config, tokenizer, compute_score, reward_router_address, reward_model_tokenizer) + + async def run_single(self, data: DataProto) -> dict: + base_result = await super().run_single(data) + return apply_format_reward_to_score(data[0], base_result) diff --git a/verl/utils/format_reward.py b/verl/utils/format_reward.py new file mode 100644 index 00000000000..e9b68973143 --- /dev/null +++ b/verl/utils/format_reward.py @@ -0,0 +1,110 @@ +# Copyright 2025 Bytedance Ltd. and/or its affiliates +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import re +from typing import Any + +import numpy as np +from verl import DataProto + +__all__ = [ + "compute_format_reward", + "apply_format_reward_to_score", + "apply_format_reward_to_tensor", +] + +# Note: +# "You can define what the 'assistant' message format should be by modifying the patterns and logic in this file. +# For example, the current implementation enforces that each assistant message must match a sequence of tags such as +# ... followed by either ... or ... (see regex patterns below). +# Adjust or extend these patterns as needed to adapt to different formatting requirements." + +def _extract_messages(messages_obj: Any) -> list[dict[str, Any]] | None: + """Normalize messages to a list of dicts.""" + if messages_obj is None: + return None + if isinstance(messages_obj, np.ndarray): + try: + messages_obj = messages_obj.item() + except ValueError: + messages_obj = messages_obj.tolist() + if isinstance(messages_obj, list): + return messages_obj + return None + + +def compute_format_reward(messages_obj: Any) -> tuple[float, dict[str, Any]]: + """Compute a format reward (+0.5 / -0.5) without attaching verbose metadata.""" + messages = _extract_messages(messages_obj) + if messages is None: + return -0.5, {} + + assistant_contents = [] + for msg in messages: + if isinstance(msg, dict) and msg.get("role") == "assistant": + assistant_contents.append(str(msg.get("content", ""))) + + if not assistant_contents: + return -0.5, {} + + # Allow 0, 1, or 2 line breaks between and the subsequent tag, + # to flexibly accept tight, single line, or double line formatting. + tool_pattern = re.compile( + r"^.*?\n{0,2}.*?$", re.DOTALL + ) + answer_pattern = re.compile( + r"^.*?\n{0,2}.*?$", re.DOTALL + ) + + for idx, content in enumerate(assistant_contents): + is_last = idx == len(assistant_contents) - 1 + expected_pattern = answer_pattern if is_last else tool_pattern + if not expected_pattern.match(content): + return -0.5, {} + + return 0.5, {} + + +def apply_format_reward_to_score(data_item: DataProto, base_result: dict) -> dict: + """Add format reward to a scalar reward score result.""" + messages = data_item.non_tensor_batch["tool_extra_fields"].get("messages") + format_reward, _ = compute_format_reward(messages) + + reward_score = base_result["reward_score"] + format_reward + reward_extra_info = dict(base_result.get("reward_extra_info", {})) + reward_extra_info["format_reward"] = format_reward + + return {"reward_score": reward_score, "reward_extra_info": reward_extra_info} + + +def apply_format_reward_to_tensor(data: DataProto, reward_tensor, reward_extra_info): + """Add format reward to the final token reward in a tensor and log extra info.""" + if "format_reward" not in reward_extra_info: + reward_extra_info["format_reward"] = [] + + for i in range(len(data)): + data_item = data[i] + prompt_ids = data_item.batch["prompts"] + prompt_length = prompt_ids.shape[-1] + valid_response_length = data_item.batch["attention_mask"][prompt_length:].sum() + + messages = data_item.non_tensor_batch["tool_extra_fields"].get("messages") + format_reward, _ = compute_format_reward(messages) + + if valid_response_length > 0: + reward_tensor[i, valid_response_length - 1] += format_reward + + reward_extra_info["format_reward"].append(format_reward) + + return reward_tensor, reward_extra_info diff --git a/verl/workers/reward_manager/__init__.py b/verl/workers/reward_manager/__init__.py index 566631b4593..19f111931d1 100644 --- a/verl/workers/reward_manager/__init__.py +++ b/verl/workers/reward_manager/__init__.py @@ -15,6 +15,8 @@ from .registry import get_reward_manager_cls, register # noqa: I001 from .batch import BatchRewardManager from .dapo import DAPORewardManager +from .format_check_dapo import FormatCheckDAPORewardManager +from .format_check_naive import FormatCheckNaiveRewardManager from .naive import NaiveRewardManager from .prime import PrimeRewardManager @@ -22,6 +24,8 @@ __all__ = [ "BatchRewardManager", "DAPORewardManager", + "FormatCheckDAPORewardManager", + "FormatCheckNaiveRewardManager", "NaiveRewardManager", "PrimeRewardManager", "register", diff --git a/verl/workers/reward_manager/format_check_dapo.py b/verl/workers/reward_manager/format_check_dapo.py new file mode 100644 index 00000000000..65cd138ffbe --- /dev/null +++ b/verl/workers/reward_manager/format_check_dapo.py @@ -0,0 +1,47 @@ +# Copyright 2025 Bytedance Ltd. and/or its affiliates +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from collections import defaultdict + +from verl.utils.format_reward import apply_format_reward_to_tensor +from verl.workers.reward_manager import register +from verl.workers.reward_manager.dapo import DAPORewardManager + + +@register("format_check_dapo") +class FormatCheckDAPORewardManager(DAPORewardManager): + """DAPO reward manager with an extra formatting bonus/penalty.""" + + def __call__(self, data, return_dict=False): + if "rm_scores" in data.batch.keys(): + if return_dict: + reward_extra_keys = data.meta_info.get("reward_extra_keys", []) + reward_extra_info = {key: data.non_tensor_batch[key] for key in reward_extra_keys} + return {"reward_tensor": data.batch["rm_scores"], "reward_extra_info": reward_extra_info} + else: + return data.batch["rm_scores"] + + base_result = super().__call__(data, return_dict=True) + reward_tensor = base_result["reward_tensor"] + + base_extra_info = base_result.get("reward_extra_info", {}) + reward_extra_info = defaultdict(list, base_extra_info) + + reward_tensor, reward_extra_info = apply_format_reward_to_tensor( + data, reward_tensor, reward_extra_info + ) + + if return_dict: + return {"reward_tensor": reward_tensor, "reward_extra_info": reward_extra_info} + return reward_tensor diff --git a/verl/workers/reward_manager/format_check_naive.py b/verl/workers/reward_manager/format_check_naive.py new file mode 100644 index 00000000000..5861aadbe1c --- /dev/null +++ b/verl/workers/reward_manager/format_check_naive.py @@ -0,0 +1,47 @@ +# Copyright 2025 Bytedance Ltd. and/or its affiliates +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from collections import defaultdict + +from verl.utils.format_reward import apply_format_reward_to_tensor +from verl.workers.reward_manager import register +from verl.workers.reward_manager.naive import NaiveRewardManager + + +@register("format_check_naive") +class FormatCheckNaiveRewardManager(NaiveRewardManager): + """Naive reward manager with an extra formatting bonus/penalty.""" + + def __call__(self, data, return_dict=False): + if "rm_scores" in data.batch.keys(): + if return_dict: + reward_extra_keys = data.meta_info.get("reward_extra_keys", []) + reward_extra_info = {key: data.non_tensor_batch[key] for key in reward_extra_keys} + return {"reward_tensor": data.batch["rm_scores"], "reward_extra_info": reward_extra_info} + else: + return data.batch["rm_scores"] + + base_result = super().__call__(data, return_dict=True) + reward_tensor = base_result["reward_tensor"] + + base_extra_info = base_result.get("reward_extra_info", {}) + reward_extra_info = defaultdict(list, base_extra_info) + + reward_tensor, reward_extra_info = apply_format_reward_to_tensor( + data, reward_tensor, reward_extra_info + ) + + if return_dict: + return {"reward_tensor": reward_tensor, "reward_extra_info": reward_extra_info} + return reward_tensor