Reinforcement Learning

pip install trinity-rft>=0.5.1
pip install flash-attn==2.8.1 --no-build-isolation

ray start --head

async def example_workflow_function(
    task: Dict,
    model: ChatModelBase,
) -> WorkflowOutput:
    agent = ReActAgent(
        name="react_agent",
        sys_prompt="You are a helpful math assistant.",
        model=model,  # use the model passed in
        formatter=OpenAIChatFormatter(),
    )

    response = await agent.reply(
        msg=Msg(
            "user",
            task["question"],  # extract question from task
            role="user",
        ),
    )

    return WorkflowOutput(  # wrap the response in WorkflowOutput
        response=response,
    )

from agentscope.tuner import tune, AlgorithmConfig, DatasetConfig, TunerModelConfig

# your workflow / judge function here...

if __name__ == "__main__":
    dataset = DatasetConfig(path="my_dataset", split="train")
    model = TunerModelConfig(model_path="Qwen/Qwen3-0.6B", max_model_len=16384)
    algorithm = AlgorithmConfig(
        algorithm_type="multi_step_grpo",
        group_size=8,
        batch_size=32,
        learning_rate=1e-6,
    )
    tune(
        workflow_func=run_react_agent,
        judge_func=judge_function,
        model=model,
        train_dataset=dataset,
        algorithm=algorithm,
    )

python main.py

current_working_directory/
    └── checkpoints/
        └──AgentScope/
            └── Experiment-20260104185355/  # each run saved in a sub-directory with timestamp
                ├── monitor/          # Monitor (e.g., Tensorboard) cache directory
                ├── log/              # tuning logs
                └── global_step_x/    # the model checkpoint saved at step x

tensorboard --logdir checkpoints/AgentScope/<your-experiment-subdir>/monitor/tensorboard

trinity convert --checkpoint-dir checkpoints/AgentScope/Experiment-20260104185355/global_step_100

async def workflow_function(
    task: Dict,
    model: ChatModelBase,
) -> WorkflowOutput:
    start_time = time.time()
    # your workflow logic here...
    end_time = time.time()
    return WorkflowOutput(
        response=response,
        metrics={
            "workflow_response_time": end_time - start_time,  # log response time for monitoring
            # you can log any other metrics you want to monitor here...
        },
    )


async def judge_function(
    task: Dict,
    response: Any,
) -> JudgeOutput:
    start_time = time.time()
    # your judge logic here...
    end_time = time.time()
    return JudgeOutput(
        reward=reward,
        metrics={
            "judge_response_time": end_time - start_time,  # log response time for monitoring
            # make sure the metric keys in workflow function and judge function are different to avoid conflicts in TensorBoard
            # you can log any other metrics you want to monitor here...
        },
    )

from logging import Logger

async def workflow_function(
    task: Dict,
    model: ChatModelBase,
    logger: Logger,
) -> WorkflowOutput:
    logger.info(f"Processing task: {task['id']}")
    # your workflow logic here...
    if error_occurred:
        logger.error(f"Error occurred while processing task: {task['id']}")
    logger.info(f"Successfully processed task: {task['id']}")
    # your workflow logic here...

path/to/log/
    ├── explorer_runner_0.log
    ├── explorer_runner_1.log
    ├── explorer_runner_2.log
    └── ...

# For TuFT, set your server address:
# export TINKER_BASE_URL=http://your-tuft-server-address
export TINKER_API_KEY=your-api-key
ray start --head

from agentscope.tuner import TunerModelConfig, TinkerConfig

model = TunerModelConfig(
    model_path="Qwen/Qwen3-4B-Instruct-2507",  # must be a model supported by the remote training API you are using
    max_model_len=16384,
    tinker_config=TinkerConfig(
        rank=16  # LoRA rank for tuning; higher values may improve performance but increase resource usage and tuning time
    )
)

checkpoint_root_dir: ./checkpoints  # Root directory for checkpoints and logs
project: AgentScope                 # Project name (sub-directory under checkpoint_root_dir)
name: Experiment-timestamp          # Experiment name (sub-directory under project)
model:
  model_path: Qwen/Qwen3-0.6B       # Model path (local or Hugging Face)
  max_model_len: 16384              # Maximum sequence length
  max_response_tokens: 4096
algorithm:
  algorithm_type: multi_step_grpo   # Tuning algorithm type (recommended: "multi_step_grpo")
  repeat_times: 8                   # Number of repeats per task (corresponds to group_size)
  optimizer:
    lr: 1e-6                        # Learning rate
buffer:
  total_epochs: 1                   # Total epochs to process the dataset
  batch_size: 32                    # Batch size per tuning iteration
  explorer_input:
    taskset:
      path: '/path/to/my_dataset'   # Dataset path (local or Hugging Face)
      split: 'train'                # Dataset split for tuning
explorer:
  rollout_model:
    engine_num: 1                   # Number of inference engines for parallel inference
    tensor_parallel_size: 1         # Tensor parallelism size
synchronizer:
  sync_style: explorer_driven
  sync_method: 'nccl'
trainer:
  save_interval: 100                # Checkpoint save interval (steps)
  ulysses_sequence_parallel_size: 1 # Ulysses sequence parallelism size
monitor:
  monitor_type: tensorboard         # Monitoring tool (tensorboard, wandb, mlflow, swanlab)
log:
  level: INFO                       # Logging level (DEBUG, INFO, WARNING, ERROR)

# Define your workflow and judge functions ...

if __name__ == "__main__":
    tune(workflow_func=workflow_function, judge_func=judge_function, config_path="config.yaml")