ART (Agent Reinforcement Trainer) - 强化学习框架训练多步任务智能体

原创

qife122

发布于 2025-07-30 21:53:47

34600

代码可运行

文章被收录于专栏：网络安全技术点滴分享网络安全技术点滴分享

运行总次数：0

代码可运行

项目标题与描述

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

采用GRPO算法进行智能体训练
创新的RULER系统自动生成奖励函数
支持真实世界复杂任务的训练场景
提供云端(SkyPilot)和本地训练选项

项目状态活跃，最新版本持续更新中。

功能特性

核心功能

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

独特价值

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

安装指南

系统要求

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

安装步骤

克隆仓库：git clone https://github.com/OpenPipe/ART.git cd ART
安装依赖：uv sync
设置环境变量：cp .env.example .env编辑.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)

典型工作流

收集轨迹：trajectories = await gather_trajectories(model, env)
训练模型：await model.train(trajectories)
评估性能：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

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

强化学习