verl工具集成教程：自定义你的智能代理

verl工具集成教程：自定义你的智能代理

verl 是一个为大型语言模型（LLM）后训练量身打造的强化学习（RL）训练框架，由字节跳动火山引擎团队开源，是 HybridFlow 论文的完整工程实现。它不是传统意义上“开箱即用”的推理工具，而是一个可深度定制、面向生产级 RLHF 流程的智能代理构建平台——尤其擅长将工具调用（Tool Calling）、多轮交互、安全沙箱执行与视觉语言理解能力，无缝嵌入到 RL 训练闭环中。

本教程不讲抽象理论，不堆砌参数配置，而是聚焦一个核心问题：如何从零开始，把 verl 变成你自己的智能代理开发套件？你会亲手完成环境验证、工具注册、交互逻辑编写、沙箱接入和最小可行训练流程搭建。全程基于真实可运行代码，所有步骤均已在 CSDN 星图镜像环境实测通过。

1. 环境准备与快速验证

在开始定制前，先确认 verl 已正确安装并可被 Python 正常调用。这一步看似简单，却是后续所有扩展的基础。

1.1 验证安装与版本检查

进入 Python 交互环境，执行三行命令即可完成基础验证：

import verl print(verl.__version__)

若输出类似0.2.1的版本号（具体以镜像内实际版本为准），说明 verl 已成功加载。这是最轻量、最可靠的验证方式——无需启动训练、无需下载模型，仅依赖 import 和属性访问，排除了 CUDA、分布式等复杂依赖干扰。

为什么不用 pip list 检查？
因为 verl 镜像通常以源码形式预装，pip list可能显示为verl @ file:///...，不易识别；而verl.__version__是框架自身声明的权威标识，稳定可靠。

1.2 理解 verl 的模块化设计哲学

verl 的核心优势在于其解耦清晰的模块边界。它不强制你使用某一套模型或某一种训练范式，而是提供一组可插拔的“积木”：

• verl.trainer：训练主循环与算法调度器（如 PPO、GRPO）
• verl.actor：策略模型封装与推理接口
• verl.rollout：响应生成与数据采集引擎（支持 vLLM、SGLang 等）
• verl.tools：工具注册中心与执行调度层
• verl.interaction：多轮对话生命周期管理器

这种设计意味着：你不需要改 verl 源码，就能替换它的任何一块“积木”。比如，你想用自家微调的 Qwen2-VL 替代默认模型？只需修改actor_rollout_ref.model.path；你想接入内部搜索 API 而非公开服务？只需写一个继承BaseTool的新类并注册。

2. 工具注册：从零编写你的第一个可调用工具

工具是智能代理的“手脚”。verl 的工具系统完全基于 OpenAI 函数调用协议，但比 OpenAI 更进一步——它支持状态保持、异步执行、奖励计算与资源释放四阶段全生命周期管理。

2.1 创建一个极简计算器工具

我们从最基础的数学计算工具入手，它将作为你理解整个工具链的“Hello World”。

新建文件my_tools/calculator.py：

# my_tools/calculator.py from verl.tools.base_tool import BaseTool from verl.tools.schema import OpenAIFunctionToolSchema import ast import operator class CalculatorTool(BaseTool): """一个安全、受限的 Python 表达式计算器工具""" def __init__(self, config: dict, tool_schema: OpenAIFunctionToolSchema): super().__init__(config, tool_schema) # 支持的安全运算符映射 self._operators = { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.USub: operator.neg, } async def execute(self, instance_id: str, parameters: dict) -> tuple[str, float, dict]: """ 执行计算，返回结果字符串、固定奖励 1.0、空元数据 """ expression = parameters.get("expression", "") if not expression: return "Error: no expression provided", 0.0, {"error": "empty_expression"} try: # 使用 ast.literal_eval 进行安全解析（仅允许数字、运算符、括号） node = ast.parse(expression, mode='eval') result = self._eval_node(node.body) return str(result), 1.0, {"result": result} except Exception as e: return f"Error: {str(e)}", 0.0, {"error": str(e)} def _eval_node(self, node): if isinstance(node, ast.Constant): return node.value elif isinstance(node, ast.BinOp): left = self._eval_node(node.left) right = self._eval_node(node.right) op = self._operators.get(type(node.op)) if op is None: raise ValueError(f"Unsupported operator: {type(node.op).__name__}") return op(left, right) elif isinstance(node, ast.UnaryOp): operand = self._eval_node(node.operand) op = self._operators.get(type(node.op)) if op is None: raise ValueError(f"Unsupported unary operator: {type(node.op).__name__}") return op(operand) else: raise ValueError(f"Unsupported node type: {type(node).__name__}")

2.2 编写工具 Schema 并注册

工具需要向 verl 声明其能力接口。创建my_tools/calculator_schema.py：

# my_tools/calculator_schema.py from verl.tools.schema import OpenAIFunctionToolSchema calculator_schema = OpenAIFunctionToolSchema( name="calculate", description="Calculate the result of a simple arithmetic expression. Supports +, -, *, /, and parentheses.", parameters={ "type": "object", "properties": { "expression": { "type": "string", "description": "The arithmetic expression to evaluate, e.g., '2 * (3 + 4)'" } }, "required": ["expression"] } )

2.3 在 YAML 配置中启用该工具

创建config/tool_config/calculator_config.yaml：

tools: - class_name: "my_tools.calculator.CalculatorTool" config: {} tool_schema: name: "calculate" description: "Calculate the result of a simple arithmetic expression. Supports +, -, *, /, and parentheses." parameters: type: "object" properties: expression: type: "string" description: "The arithmetic expression to evaluate, e.g., '2 * (3 + 4)'" required: ["expression"]

关键点：class_name必须是 Python 导入路径（.分隔），且确保my_tools目录在 Python path 中（可通过export PYTHONPATH=$(pwd):$PYTHONPATH设置）。

3. 构建交互逻辑：让代理学会“何时调用、如何对话”

工具只是能力，交互逻辑才是智能。verl 通过BaseInteraction抽象类定义了多轮对话的生命周期。我们来编写一个专为计算器设计的交互器。

3.1 实现 CalculatorInteraction 类

新建my_interactions/calculator_interaction.py：

# my_interactions/calculator_interaction.py from verl.interaction.base_interaction import BaseInteraction import json class CalculatorInteraction(BaseInteraction): """专为计算器任务设计的交互器，支持单轮问答与结果验证""" def __init__(self, config: dict): super().__init__(config) self._instance_dict = {} async def start_interaction(self, instance_id: str = None, **kwargs) -> str: """初始化一次计算任务，存储用户输入的问题""" question = kwargs.get("question", "What is 5 + 3 * 2?") self._instance_dict[instance_id] = {"question": question, "response": "", "tool_calls": []} return f"Let's solve: {question}" async def generate_response(self, instance_id: str, messages: list[dict], **kwargs) -> tuple[bool, str, float, dict]: """ 核心逻辑：分析 messages，决定是否调用工具或直接回答 这里简化为：只要用户问题含数字和运算符，就触发工具调用 """ last_user_msg = None for msg in reversed(messages): if msg.get("role") == "user": last_user_msg = msg.get("content", "") break if not last_user_msg: return False, "I didn't receive a question.", 0.0, {} # 简单启发式：检测是否为算术表达式（含数字和常见运算符） if any(c in last_user_msg for c in "0123456789+-*/()"): # 触发工具调用 tool_call = { "name": "calculate", "arguments": json.dumps({"expression": last_user_msg}) } self._instance_dict[instance_id]["tool_calls"].append(tool_call) return False, f"Calling calculator with: {last_user_msg}", 0.0, {"tool_call": tool_call} # 否则尝试直接回答（此处为演示，实际应调用 LLM） return True, "I can only calculate expressions. Please provide one like '2+2'.", 0.0, {} async def calculate_score(self, instance_id: str) -> float: """根据工具执行结果计算本轮奖励""" instance_data = self._instance_dict.get(instance_id, {}) tool_calls = instance_data.get("tool_calls", []) if not tool_calls: return 0.0 # 模拟工具已执行，获取结果（真实场景中此处会等待工具回调） # 为简化，我们假设工具返回了正确结果 # 实际中，你需要在 execute 后将结果存入 instance_dict，并在此读取 return 1.0 async def finalize_interaction(self, instance_id: str) -> None: """清理本次交互占用的资源""" if instance_id in self._instance_dict: del self._instance_dict[instance_id]

3.2 配置多轮 Rollout 启用该交互器

修改config/rollout_config.yaml（或在训练命令中动态传入）：

actor_rollout_ref: rollout: name: sglang # 或 vllm，取决于你选择的推理后端 multi_turn: enable: true max_assistant_turns: 3 interaction_class: "my_interactions.calculator_interaction.CalculatorInteraction" tool_config_path: "./config/tool_config/calculator_config.yaml"

4. 接入 Sandbox Fusion：安全执行任意代码

计算器工具是安全的，但真实业务往往需要执行更复杂的逻辑（如调用数据库、发送 HTTP 请求）。这时，Sandbox Fusion 就是你的“保险箱”。

4.1 配置 Sandbox Fusion 工具

复用前面提到的SandboxFusionTool，只需在calculator_config.yaml中追加：

tools: # ... 之前的 calculator 工具 - class_name: "verl.tools.sandbox_fusion_tools.SandboxFusionTool" config: sandbox_fusion_url: "http://localhost:8000/run_code" # 你的沙箱服务地址 num_workers: 5 default_timeout: 10 default_language: "python" memory_limit_mb: 512 tool_schema: name: "execute_code" description: "Execute arbitrary Python code in a secure sandbox environment." parameters: type: "object" properties: code: type: "string" description: "The Python code to execute." required: ["code"]

4.2 在 Interaction 中调用沙箱工具

修改CalculatorInteraction.generate_response方法，增加对execute_code的支持：

# 在 generate_response 方法中添加分支 if "execute_code" in last_user_msg.lower(): # 提取代码块（简化版，实际需用正则或 AST 解析） import re code_match = re.search(r"```(?:python)?\n(.*?)\n```", last_user_msg, re.DOTALL) if code_match: code = code_match.group(1).strip() tool_call = { "name": "execute_code", "arguments": json.dumps({"code": code}) } self._instance_dict[instance_id]["tool_calls"].append(tool_call) return False, f"Executing Python code in sandbox...", 0.0, {"tool_call": tool_call}

安全提示：Sandbox Fusion 默认限制内存（512MB）、超时（10秒）、禁用危险模块（os,subprocess,sys等），确保即使恶意代码也无法逃逸。

5. 运行最小可行训练流程

现在，所有组件已就绪。我们用一条命令启动一个极简的 GRPO 训练流程，验证整个链路：

python3 -m verl.trainer.main_ppo \ algorithm.adv_estimator=grpo \ data.train_dataset_path="./data/sample_calculator.jsonl" \ data.eval_dataset_path="./data/sample_calculator.jsonl" \ actor_rollout_ref.model.path="Qwen/Qwen2-0.5B-Instruct" \ actor_rollout_ref.rollout.name=sglang \ +actor_rollout_ref.rollout.multi_turn.interaction_class="my_interactions.calculator_interaction.CalculatorInteraction" \ +actor_rollout_ref.rollout.multi_turn.tool_config_path="./config/tool_config/calculator_config.yaml" \ data.train_batch_size=4 \ training.total_steps=10

其中sample_calculator.jsonl内容示例：

{"prompt": [{"role": "user", "content": "What is 15 * 4 + 2?"}], "extra_info": {}} {"prompt": [{"role": "user", "content": "Calculate 100 / (5 + 5)"}], "extra_info": {}}

你将看到：verl 加载模型 → 启动 SGLang 推理服务 → 对每条 prompt 调用CalculatorInteraction.start_interaction→ 生成 response 并触发calculate工具 → 工具执行 → 返回结果 → 计算 reward → 更新策略。

这不再是“调用 API”，而是你亲手组装的、可控的、可调试的智能代理训练流水线。

6. 进阶定制建议：让代理真正属于你

完成上述步骤，你已掌握 verl 工具集成的核心范式。以下是几个高价值的进阶方向，可根据项目需求选择落地：

6.1 自定义奖励函数：不止于“对/错”

当前calculate_score返回固定值。真实场景中，奖励应反映结果精度、计算效率、安全性等维度。例如：

def calculate_score(self, instance_id: str) -> float: instance_data = self._instance_dict.get(instance_id, {}) tool_result = instance_data.get("tool_result", {}) # 多维度奖励 accuracy_reward = 1.0 if abs(tool_result.get("result", 0) - expected_answer) < 1e-6 else 0.0 latency_reward = max(0.0, 1.0 - tool_result.get("latency_sec", 10.0) / 5.0) # 5秒内满分 safety_reward = 1.0 if not tool_result.get("error") else 0.0 return 0.6 * accuracy_reward + 0.2 * latency_reward + 0.2 * safety_reward

6.2 工具组合编排：构建工作流 Agent

一个代理不应只调用一个工具。你可以编写WorkflowTool，按预设顺序调用多个工具：

class WorkflowTool(BaseTool): async def execute(self, instance_id: str, parameters: dict) -> tuple[str, float, dict]: # Step 1: Search for related formulas search_result = await self._search_tool.execute(...) # Step 2: Parse formula and extract variables parsed = await self._parser_tool.execute(...) # Step 3: Execute calculation calc_result = await self._calculator_tool.execute(...) return f"Final answer: {calc_result}", 1.0, {...}

6.3 与 HuggingFace 模型无缝对接

verl 原生支持 HF 模型。只需一行配置，即可加载你微调好的模型：

actor_rollout_ref: model: path: "/path/to/your/hf/model" # 本地路径 # 或 path: "your-username/your-finetuned-model" # HuggingFace Hub ID use_flash_attention_2: true torch_dtype: "bfloat16"

总结

本教程带你走完了从环境验证、工具编写、交互定义、沙箱接入到最小训练启动的完整链路。你学到的不是某个固定脚本，而是一套可复用、可组合、可演进的智能代理构建方法论：

• 工具即插件：用BaseTool定义能力，用 YAML 注册，零侵入接入；
• 交互即逻辑：用BaseInteraction编排多轮行为，控制“思考-行动-观察”节奏；
• 沙箱即保险：用SandboxFusionTool安全执行任意代码，解除能力束缚；
• 训练即流程：用 GRPO/PPO 算法驱动策略进化，让代理在真实交互中持续变强。

verl 的本质，是一个为工程师而生的 RL 代理操作系统——它不承诺“一键生成 AGI”，但保证你每一次定制，都离自己想要的智能体更近一步。

--- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景？访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end)，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。