[rollout,vllm] fix: num_preempted metrics fix and typo correction in vllm async server (verl-project#4976)

### What does this PR do?

fix the num_preempted metrics in multi-turn,and the typo in
vllm_async_server

### Checklist Before Starting

- [x] Search for similar PRs. Paste at least one query link here: ...
- [x] Format the PR title as `[{modules}] {type}: {description}` (This
will be checked by the CI)
- `{modules}` include `fsdp`, `megatron`, `veomni`, `sglang`, `vllm`,
`rollout`, `trainer`, `ci`, `training_utils`, `recipe`, `hardware`,
`deployment`, `ray`, `worker`, `single_controller`, `misc`, `perf`,
`model`, `algo`, `env`, `tool`, `ckpt`, `doc`, `data`, `cfg`, `reward`
- If this PR involves multiple modules, separate them with `,` like
`[megatron, fsdp, doc]`
  - `{type}` is in `feat`, `fix`, `refactor`, `chore`, `test`
- If this PR breaks any API (CLI arguments, config, function signature,
etc.), add `[BREAKING]` to the beginning of the title.
  - Example: `[BREAKING][fsdp, megatron] feat: dynamic batching`

### Test

> For changes that can not be tested by CI (e.g., algorithm
implementation, new model support), validate by experiment(s) and show
results like training curve plots, evaluation results, etc.

### API and Usage Example

> Demonstrate how the API changes if any, and provide usage example(s)
if possible.

```python
# Add code snippet or script demonstrating how to use this
```

### Design & Code Changes

> Demonstrate the high-level design if this PR is complex, and list the
specific changes.

### Checklist Before Submitting

> [!IMPORTANT]
> Please check all the following items before requesting a review,
otherwise the reviewer might deprioritize this PR for review.

- [ ] Read the [Contribute
Guide](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md).
- [ ] Apply [pre-commit
checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting):
`pre-commit install && pre-commit run --all-files --show-diff-on-failure
--color=always`
- [ ] Add / Update [the
documentation](https://github.com/volcengine/verl/tree/main/docs).
- [ ] Add unit or end-to-end test(s) to [the CI
workflow](https://github.com/volcengine/verl/tree/main/.github/workflows)
to cover all the code. If not feasible, explain why: ...
- [ ] Once your PR is ready for CI, send a message in [the `ci-request`
channel](https://verl-project.slack.com/archives/C091TCESWB1) in [the
`verl` Slack
workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ).
(If not accessible, please try [the Feishu group
(飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).)
- [ ] If your PR is related to the `recipe` submodule, please also
update the reference to the submodule commit via `git submodule update
--remote` or `cd recipe && git pull origin main`.

Author: RobotGF

verl/experimental/agent_loop/agent_loop.py

@@ -983,6 +983,7 @@ def _performance_metrics(self, metrics: list[list[dict[str, str]]], output: Data
         timing["agent_loop/slowest/tool_calls"] = t_tool_calls[slowest]
         timing["agent_loop/slowest/prompt_length"] = attention_mask[:prompt_length].sum().item()
         timing["agent_loop/slowest/response_length"] = attention_mask[prompt_length:].sum().item()
+        timing["agent_loop/slowest/num_preempted"] = num_preempted[slowest]
 
         return timing
 

verl/experimental/agent_loop/single_turn_agent_loop.py

@@ -63,8 +63,8 @@ async def run(self, sampling_params: dict[str, Any], **kwargs) -> AgentLoopOutpu
                 image_data=images,
                 video_data=videos,
             )
-        if output.num_preempted is not None:
-            metrics["num_preempted"] = output.num_preempted
+        if metrics.get("num_preempted") is None:
+            metrics["num_preempted"] = output.num_preempted if output.num_preempted is not None else -1
         response_mask = [1] * len(output.token_ids)
 
         output = AgentLoopOutput(

verl/experimental/agent_loop/tool_agent_loop.py

@@ -233,8 +233,12 @@ async def _handle_generating_state(
                 image_data=agent_data.image_data,
                 video_data=agent_data.video_data,
             )
-        if output.num_preempted is not None:
-            agent_data.metrics["num_preempted"] = output.num_preempted
+        # first time to set num_preempted
+        if agent_data.metrics.get("num_preempted") is None:
+            agent_data.metrics["num_preempted"] = output.num_preempted if output.num_preempted is not None else -1
+        # then add num_preempted to the metrics
+        else:
+            agent_data.metrics["num_preempted"] += output.num_preempted if output.num_preempted is not None else 0
 
         agent_data.assistant_turns += 1
         agent_data.response_ids = output.token_ids

verl/workers/rollout/vllm_rollout/vllm_async_server.py

@@ -561,7 +561,7 @@ async def generate(
         num_preempted = None
 
         if hasattr(final_res.outputs[0], "num_preempted"):
-            num_preempted = final_res.output[0].num_preempted
+            num_preempted = final_res.outputs[0].num_preempted
 
         return TokenOutput(
             token_ids=token_ids,