项目标题与描述

ART (Agent Reinforcement Trainer) 是一个开源的强化学习框架，专注于训练能够处理多步任务的智能体。项目核心价值在于：

  • 采用GRPO算法进行智能体训练
    • 创新的RULER系统自动生成奖励函数
    • 支持真实世界复杂任务的训练场景
    • 提供云端(SkyPilot)和本地训练选项 项目状态活跃，最新版本持续更新中。

功能特性

核心功能

  • RULER自动奖励系统：
    • 使用LLM作为评判者自动评分轨迹
    • 无需手工设计奖励函数
    • 支持多任务通用评分
  • 灵活的训练部署：
    • 支持云端集群训练(SkyPilot)
    • 本地开发环境支持
    • VSCode远程开发集成
  • 多领域任务支持：
    • 零售领域任务(订单处理、退换货等)
    • 航空领域任务(机票预订、改签等)
    • 自定义任务扩展
  • 强化学习工具链：
    • 轨迹收集与评估
    • 模型训练与部署
    • 性能监控(W&B集成)

独特价值

  • 开发效率提升：相比传统RL方法节省2-3倍开发时间
    • 通用性强：同一套系统可应用于不同领域任务
    • 性能优越：在基准测试中超越手工设计奖励函数

安装指南

系统要求

  • Python 3.8+
    • CUDA 11.7+ (GPU训练需要)
    • UV包管理器(推荐)

安装步骤

  1. 克隆仓库：
  3. git clone https://github.com/OpenPipe/ART.git
  4. cd ART
  6. 安装依赖：
  8. uv sync
  10. 设置环境变量：
  12. cp .env.example .env

  13. 编辑.env文件配置您的设置

可选组件

  • 树形解析器支持：
  • pip install 'tree-sitter==0.21.3'
  • pip install 'tree-sitter-languages'

使用说明

基础示例

启动训练集群(SkyPilot)：

./scripts/launch-cluster.sh

运行2048示例：

import art
model = art.TrainableModel(name="qwen2.5-14b-instruct",project="demo",base_model="Qwen/Qwen2.5-14B-Instruct"
)
await model.register(backend)

典型工作流

  1. 收集轨迹：
  3. trajectories = await gather_trajectories(model, env)
  5. 训练模型：
  7. await model.train(trajectories)
  9. 评估性能：
  11. results = await evaluate_model(model, test_env)

API概览

主要组件：

  • TrainableModel: 可训练模型封装
    • LocalBackend: 本地训练后端
    • Trajectory: 智能体轨迹数据类
    • RULER: 自动奖励评分系统

核心代码

RULER自动评分系统

async def ruler_score_group(group, model_name):"""使用LLM自动评分轨迹组"""prompt = build_scoring_prompt(group.trajectories)response = await llm_completion(prompt, model_name)return parse_scores(response)def build_scoring_prompt(trajectories):"""构建评分提示"""return f"""评估以下轨迹，根据任务完成度打分(0-1):{json.dumps([t.to_dict() for t in trajectories])}"""

智能体训练循环

async def train_loop(model, env, steps=1000):for step in range(steps):# 收集轨迹trajectories = await gather_trajectories(model, env)# 使用RULER评分scored = await ruler_score_group(trajectories, "o3")# 训练步骤await model.train(scored)# 定期评估if step % 10 == 0:eval_results = await evaluate(model, test_env)log_to_wandb(eval_results)

云端部署

async def deploy_to_cloud(model, step):"""部署模型到云端推理服务"""deployment = await deploy_model(deploy_to="together",model=model,step=step,wait_for_completion=True)return deployment.endpoint

更多精彩内容 请关注我的个人公众号 公众号（办公AI智能小助手）