[sglang] feat: Support async multi-turn rollout with simulation feedback in sglang

[kinza99]

Checklist Before Starting

• Searched for similar PR(s).
• Checked PR Title format
  • In format of: [modules] type: Title
  • modules are in fsdp, megatron, sglang, vllm, rollout, trainer, ci, training_utils, recipe, hardware, deployment, ray, worker, single_controller, misc, perf, model, algo, env, tool, ckpt, doc, data
  • type is in feat, fix, refactor, chore, test
  • can involve multiple modules, seperated by , or space, like [megatron, fsdp, doc] feat: xxx

What does this PR do?

Implements multi-turn interaction system for reinforcement learning training, enabling dynamic conversational feedback and iterative problem-solving scenarios with extensible agent-based architecture.

Test

Comprehensive testing has been performed including:

• Unit Tests: Complete test suite in tests/interactions/test_gsm8k_interaction.py with 15+ test cases covering:
  • Initialization and configuration validation
  • Interaction lifecycle (start → generate → calculate → finalize)
  • Correct/incorrect answer handling with GSM8K scoring integration
  • Edge cases (empty messages, malformed content, concurrent sessions)
  • Resource cleanup and error handling
• Integration Tests: SGLang rollout integration in tests/workers/rollout/test_sglang_async_rollout_w_interaction.py:
  • Multi-GPU distributed testing (requires 2+ GPUs)
  • FSDP model sharding with SGLang inference engine
  • End-to-end interaction workflow with interaction_kwargs parameter passing
  • Comparison with HuggingFace baseline for output validation
• Real-world Validation: Training script testing with Qwen2.5-0.5B model demonstrating:
  • GRPO algorithm integration with interaction-based rewards
  • Multi-turn conversation handling in production training loops

High-Level Design

Multi-Turn Interaction System Architecture:

The system introduces a flexible, async-based interaction framework designed for RL training scenarios with the following key components:

1. BaseInteraction Class: Core abstraction layer providing async interface for interaction agents
2. Instance Management: Stateful session management with unique instance IDs for concurrent interactions
3. SGLang Integration: Seamless integration with SGLang rollout system for multi-turn conversations
4. Configuration-Driven Loading: Dynamic agent loading via YAML configuration files
5. Reward Integration: Turn-level scoring mechanism integrated with VERL's reward system

Specific Changes

Core Implementation:

• Added BaseInteraction abstract class with async interface in verl/interactions/base.py
• Implemented Gsm8kInteraction concrete class for math problem solving scenarios in verl/interactions/gsm8k_interaction.py
• Added instance-based state management for concurrent interaction sessions via _instance_dict
• Created turn-level scoring mechanism using verl.utils.reward_score.gsm8k with flexible answer extraction

SGLang Rollout Integration (verl/workers/rollout/sglang_rollout/sglang_rollout.py):

• Extended _async_rollout_a_request method with AsyncRolloutRequestStateEnum.INTERACTING state
• Added interaction initialization in _handle_pending_state method (line 878-880)
• Implemented interaction response generation with reward accumulation (line 823-835)
• Dynamic interaction loading via _intitalize_interaction method with importlib-based class resolution
• Configuration-driven setup supporting interaction_config_path parameter

Testing Infrastructure:

• Comprehensive unit test suite in tests/interactions/test_gsm8k_interaction.py with pytest and asyncio support
• Integration tests in tests/workers/rollout/test_sglang_async_rollout_w_interaction.py with distributed GPU testing
• Mock-based testing using unittest.mock.patch for GSM8K scoring validation

Configuration & Examples:

• GSM8K interaction configuration in examples/sglang_multiturn/config/interaction_config/gsm8k_interaction_config.yaml
• Training script: run_qwen2.5-0.5b_gsm8k_multiturn_w_interaction.sh with interaction_config_path parameter
• Support for max_user_turns and max_assistant_turns configuration

API

BaseInteraction Interface:

from verl.interactions.base import BaseInteraction
from typing import Dict, Any, List, Tuple, Optional

class CustomInteraction(BaseInteraction):
    def __init__(self, config: Dict[str, Any]):
        super().__init__(config)
    
    async def start_interaction(self, instance_id: Optional[str] = None, **kwargs) -> str:
        # Initialize interaction session, return instance_id
        pass
    
    async def generate_response(self, instance_id: str, messages: List[Dict[str, Any]], **kwargs) -> Tuple[bool, str, float, Dict[str, Any]]:
        # Generate response, return (should_terminate, response, score, metadata)
        pass
    
    async def calculate_score(self) -> float:
        # Calculate turn-level score for RL training
        pass
        
    async def finalize_interaction(self) -> None:
        # Clean up resources
        pass

Usage Example

GSM8K Interaction Configuration:

# gsm8k_interaction_config.yaml
interaction:
  - class_name: "verl.interactions.gsm8k_interaction.Gsm8kInteraction"
    config: {}

Training Script Integration:

python3 -m verl.trainer.main_ppo \
    --config-path="$CONFIG_PATH" \
    --config-name='gsm8k_multiturn_grpo_w_interaction' \
    algorithm.adv_estimator=grpo \
    data.train_batch_size=512 \
    data.return_raw_chat=True \
    actor_rollout_ref.rollout.name=sglang \
    actor_rollout_ref.rollout.multi_turn.interaction_config_path="$PROJECT_DIR/examples/sglang_multiturn/config/interaction_config/gsm8k_interaction_config.yaml" \
    trainer.total_epochs=15

GSM8K Interaction Implementation:

class Gsm8kInteraction(BaseInteraction):
    def __init__(self, config: dict):
        super().__init__(config)
        self._instance_dict = {}

    async def start_interaction(self, instance_id=None, ground_truth=None, **kwargs):
        if instance_id is None:
            instance_id = str(uuid4())
        self._instance_dict[instance_id] = {
            "response": "",
            "ground_truth": ground_truth,
            "reward": 0.0,
        }
        return instance_id

    async def generate_response(self, instance_id, messages, **kwargs):
        # Extract last user message content
        content = ""
        for item in reversed(messages):
            if item.get("role") == "user":
                content = item.get("content", "")
                break

        # Ensure GSM8K format (#### prefix)
        if content.startswith("#### "):
            self._instance_dict[instance_id]["response"] = content
        else:
            self._instance_dict[instance_id]["response"] = "#### " + content

        reward = await self.calculate_score(instance_id)
        if reward == 1.0:
            return True, "Your response is correct!", 1.0, {}
        else:
            return False, "Your response is incorrect! You need to reflect on your answer and try again.", 0.0, {}

    async def calculate_score(self, instance_id, **kwargs):
        return gsm8k.compute_score(
            self._instance_dict[instance_id]["response"],
            self._instance_dict[instance_id]["ground_truth"],
            method="flexible", format_score=0.0, score=1.0,
        )

Contributor List

• He Du (Author)
• Xiang Long (Co-author)
• Yanbin Jiang (Discussed integrated with LangGraph scene)
• Junrong Lin (Reviewer)
• Haibin Lin (Reviewer)
• Chenyang Zhao (PM)

Checklist Before Submitting

• Read the Contribute Guide.
• Apply pre-commit checks.
• Add [BREAKING] to the PR title description if it breaks any API.
• Update the documentation about your changes in the docs.
• New CI unit test(s) are added to cover the code path.
• Rely on existing unit tests on CI that covers the code path.

[CLAassistant]

All committers have signed the CLA.

[kinza99]

@zhaochenyang20

[zhaochenyang20]

nice work！

[zhaochenyang20]

We will take a look these days, stay tuned in 24h
.

[zhaochenyang20]

@kinza99 my wechat is 18015766633. Feel free to discuss, thanks!

[SwordFaith]

It might be better to rename it to "interaction_kwargs".

[SwordFaith]

interaction_config_file

[SwordFaith]

It might be better support uuid by default like https://github.com/volcengine/verl/blob/main/verl/tools/base_tool.py#L50.

[SwordFaith]

Sorry for missing instance_id in doc, it would be better keep instance id logic in tool to track session.

[SwordFaith]

feedback -> response

[SwordFaith]

WAITING is a bad state name, is there any other concrete state name for interacting with Interaction cls?

[SwordFaith]

Need add unit test for new added interaction related logic.

[ocss884]

Just curious why adding lot .keys() in this PR? For ... in dict_like_obj the statements are equivalent

[kinza99]

In some versions, you may encounter the problem NotImplementedError: TensorDict does not support membership checks with the in keyword. If you want to check if a particular key is in your TensorDict, please use key in tensordict.keys() instead.

[ocss884]

How about raise NotImplementedError

[SwordFaith]

How about raise NotImplementedError

The calc_score method is not essential for implementing interactions. Here, we use 0.0 to ensure it has no impact on the aggregated reward.

[ocss884]

Duplicate torch.distributed

[ocss884]

pls add the copyright of bytedance

[SwordFaith]

To-do:

1. Review new examples in wandb log.
2. Refactor duplicated unit tests.

[eric-haibin-lin]

There seems to be lots of changes. I'll take some time to review as well

[vermouth1992]

Is this just an OpenAI GymEnv?

[SwordFaith]

I guess env or tool can both support in current tools submodule. Interaction is designed to 2 sceneraios:

• partial exposure
• inject context with workflow as user role (as many framework do)

So BaseInteraction support generate response from model api our other workflow part as endpoint. In partial exposure user requirements, typically need message history to dynamicly add user requirements. In workflow context injection, it typically need messages parsed to other node input and inject other node output back to current node. And we can use e2e outcome to train current node model (If not in current model, need to extend in finalize).

[wuxibin89]

P2: use a pydantic class as return instead of Tuple

[SwordFaith]

Will switch to pydantic cls in future.

[wuxibin89]

typo: self.interaction is not a dict

[SwordFaith]

It could be fixed in _initialize_interaction, which should support init multiple interactions in same config file. I guess we coulld merge 1630 first and add multiple interactions in future. Current impl already support single task training with interaction.

[wuxibin89]

Do we support choose different interaction based on assistant response? Or we just use one interaction, and dispatch in it?

[SwordFaith]

We support different interaction cls, according to sample level config, like tools. For message history difference,should be handled inside interaction.

[PeterSH6]

Could you add a config to control whether to include the tool related fields in dataproto? This may increase the size if tools are not used

[SwordFaith]

#2145 Tracked by new issue, will refactor in future.

[leoleoasd]

Why is user message used to calcualte the reward instead of the assistant message?

[SwordFaith]

Why is user message used to calcualte the reward instead of the assistant message?

We are developing a message and turn-level reward system. Calculating rewards during the user's turn doesn't necessarily mean that the user's message requires a reward. Instead, it allows the user to reallocate rewards between the assistant's turns based on the user's turn rewards.