在`rollout` 命令中使用参数`use_async_engine` 来指定 engine 的种类（默认使用 async engine）：

在训练侧设置参数`--vllm_server_pass_dataset`，可将数据集中的其他列传入多轮规划器。在`infer_request.data_dict`中获取。

swift >= 3.11 支持从 vLLM 侧返回 rollouot 的 logps 用于纠正训推不一致问题，具体请参考该[文档](../AdvancedResearch/training_inference_mismatch.md)

在多轮训练中，如果启用了`rollout_importance_sampling_mode`，框架会自动收集每轮 rollout 的 log probabilities，用于校正训推不一致带来的 off-policy 问题。

**默认行为**：

- 使用默认的`run` 方法时，框架会自动从`response_choice.logprobs` 中提取 log probabilities

- 这些 logprobs 会与`response_token_ids` 和`response_loss_mask` 一起传递给 trainer

**自定义 Scheduler 的注意事项**：

如果你在`step` 方法中修改了 response（如截断、添加内容），需要同步返回对应的`rollout_logprobs`

**关键规则**：

-`rollout_logprobs` 的长度应该等于`response_loss_mask` 中值为 1 的数量

- 对于`loss_mask=0` 的 token（如用户添加的提示、工具返回结果），不需要提供 logprobs

- 如果`step` 方法没有返回`rollout_logprobs`，框架会自动从`response_choice.logprobs` 中提取

**重写`run` 方法的场景**：

如果你完全重写了`run` 方法，需要手动收集和传递`rollout_logprobs`

具体的实现请参考[内置实现](https://github.com/modelscope/ms-swift/blob/main/swift/plugin/multi_turn.py)

Use the`use_async_engine` argument in the`rollout` command to specify the engine type (async is the default).

Set`--vllm_server_pass_dataset` on the training side to pass other dataset columns to the scheduler.

They can be read from`infer_request.data_dict`.

Swift >= 3.11 supports returning rollout logprobs from the vLLM side to address training-inference mismatch issues. For details, please refer to this[document](../AdvancedResearch/training_inference_mismatch.md).

In multi-turn training, if`rollout_importance_sampling_mode` is enabled, the framework automatically collects log probabilities from each rollout turn to correct off-policy issues.

**Default Behavior**:

- When using the default`run` method, the framework automatically extracts log probabilities from`response_choice.logprobs`

- These logprobs are passed to the trainer along with`response_token_ids` and`response_loss_mask`

**Notes for Custom Schedulers**:

If you modify the response in your`step` method (e.g., truncation, adding content), you need to return the corresponding`rollout_logprobs`:

**Key Rules**:

- The length of`rollout_logprobs` should equal the count of 1s in`response_loss_mask`

- For tokens with`loss_mask=0` (e.g., user-added prompts, tool return results), no logprobs are needed

- If`step` does not return`rollout_logprobs`, the framework will automatically extract them from`response_choice.logprobs`

**When Overriding the`run` Method**:

If you completely override the`run` method, you need to manually collect and pass`rollout_logprobs`

For implementation, please refer to[here](https://github.com/modelscope/ms-swift/blob/main/swift/plugin/multi_turn.py)

response_token_ids:List[List[int]]=Field(default_factory=list)

response_loss_mask:List[List[int]]=Field(default_factory=list)

rollout_infos:Dict[str,Any]=Field(default_factory=dict)

rollout_logprobs:List[List[float]]=Field(default_factory=list)

def_wrap_flat_list(cls,v):

ifisinstance(v,list)andvandisinstance(v[0],int):

ifisinstance(v,list)andvandisinstance(v[0],(int,float)):

return [v]

*args,

**kwargs):

self.infer_engine=infer_engine

self._tokenizer=kwargs.get('tokenizer',None)

self.max_turns=max_turns

asyncdefasync_infer(self,

defengine(self):

returnself.infer_engine

deftokenizer(self):

ifself._tokenizerisnotNone:

returnself._tokenizer

ifself.infer_engineisnotNone:

returnself.infer_engine.tokenizer

classMultiTurnScheduler(RolloutScheduler,ABC):

rollout_infos= {}

total_response_ids= []

total_response_loss_mask= []

total_rollout_logprobs= []

whileTrue:

messages=current_request.messages

ifcurrent_turn==1ornotmessages[-1]['content']:

should_stop=should_stopor (current_turn>=self.max_turns)

ifshould_stop:

current_logprobs=self._extract_logprobs_from_choice(response_choice)

final_token_ids=response_choice.token_ids

ifis_continuationandtotal_response_ids:

total_response_ids[-1].extend(response_choice.token_ids)

total_response_ids[-1].extend(final_token_ids)

iftotal_response_loss_mask:

total_response_loss_mask[-1].extend([1]*len(response_choice.token_ids))

total_response_loss_mask[-1].extend([1]*len(final_token_ids))

iftotal_rollout_logprobsandcurrent_logprobs:

total_rollout_logprobs[-1].extend(current_logprobs)

elifnottotal_response_ids:

iffinal_token_ids:

total_response_ids= [list(final_token_ids)]

total_response_loss_mask= [[1]*len(final_token_ids)]

ifcurrent_logprobs:

total_rollout_logprobs= [current_logprobs]

iftotal_rollout_logprobs:

total_logprob_count=sum(len(turn_lps)forturn_lpsintotal_rollout_logprobs)

iftotal_response_loss_mask:

total_loss_mask_1_count=sum(sum(mask)formaskintotal_response_loss_mask)

iftotal_loss_mask_1_count!=total_logprob_count:

final_rollout_logprobs= []

else:

iftotal_response_ids:

total_response_id_count=sum(len(turn_ids)forturn_idsintotal_response_ids)

iftotal_response_id_count!=total_logprob_count:

final_rollout_logprobs= []

else:

final_rollout_logprobs= []

returnRolloutOutput(

response=response,

rollout_infos={

**rollout_infos,'num_turns':current_turn

},

rollout_logprobs=final_rollout_logprobs,

)

rollout_infos.update(ret['rollout_infos'])

if'rollout_logprobs'inretandret['rollout_logprobs']:

current_logprobs=ret['rollout_logprobs']

else:

current_logprobs=self._extract_logprobs_from_choice(response_choice)

ifcurrent_logprobs:

ifis_continuationandtotal_rollout_logprobs:

total_rollout_logprobs[-1].extend(current_logprobs)

else:

total_rollout_logprobs.append(current_logprobs)

ifcurrent_request.messages[-1]['role']=='assistant':

current_request.messages.append({'role':'assistant','content':None})

def_extract_logprobs_from_choice(response_choice:'ChatCompletionResponseChoice')->List[float]:

ifresponse_choice.logprobsisNone:

return []

if'content'inresponse_choice.logprobs:

return [item['logprob']foriteminresponse_choice.logprobs['content']]

return []

classThinkingModelTipsScheduler(MultiTurnScheduler):

from .ormimportMathAccuracy

super().__init__(*args,**kwargs)

self.acc_func=kwargs.get('acc_function',MathAccuracy())

self._tips_token_ids=None

def_get_tips_token_ids(self,tokenizer)->List[int]:

ifself._tips_token_idsisNone:

self._tips_token_ids=tokenizer.encode(self.tips_prompt,add_special_tokens=False)

returnself._tips_token_ids

defcheck_finished(self,infer_request:'RolloutInferRequest',response_choice:'ChatCompletionResponseChoice',

current_turn:int)->bool:

defstep(self,infer_request:'RolloutInferRequest',response_choice:'ChatCompletionResponseChoice',

current_turn:int)->Dict:

completion=response_choice.message.content

response_token_ids=list(response_choice.token_ids)ifresponse_choice.token_idselse []

rollout_logprobs=self._extract_logprobs_from_choice(response_choice)

truncate_idx=len(completion)

if'<answer>'incompletion:

completion=completion[:completion.index('<answer>')]

truncate_idx=min(truncate_idx,completion.index('<answer>'))

if'</think>'incompletion:

completion=completion[:completion.index('</think>')]

truncate_idx=min(truncate_idx,completion.index('</think>'))

iftruncate_idx<len(completion):

truncated_completion=completion[:truncate_idx]

ifresponse_token_idsandself.tokenizerisnotNone:

token_truncate_idx=len(response_token_ids)

foriinrange(1,len(response_token_ids)+1):

decoded=self.tokenizer.decode(response_token_ids[:i],skip_special_tokens=False)

iflen(decoded)>=truncate_idx:

response_token_ids=response_token_ids[:token_truncate_idx]

ifrollout_logprobs:

rollout_logprobs=rollout_logprobs[:token_truncate_idx]

completion+=self.tips_prompt

ifresponse_token_idsandself.tokenizerisnotNone:

tips_token_ids=self._get_tips_token_ids(self.tokenizer)

response_loss_mask= [1]*len(response_token_ids)+ [0]*len(tips_token_ids)

else:

response_loss_mask= []

ifinfer_request.messages[-1]['role']=='assistant':

ifnotinfer_request.messages[-1]['content']:

else:

infer_request.messages.append({'role':'assistant','content':completion})

return {'infer_request':infer_request}

result= {'infer_request':infer_request}

ifresponse_token_ids:

result['response_token_ids']=response_token_ids

result['response_loss_mask']=response_loss_mask

ifrollout_logprobs:

result['rollout_logprobs']=rollout_logprobs

classGYMScheduler(RolloutScheduler):