verl多工具协同实战：搜索+计算一体化

1. 引言：当强化学习遇上真实世界任务

你有没有想过，一个AI模型不仅能写文章、做推理，还能主动上网查资料、运行代码、验证结果？这不再是科幻场景。借助verl——这个由字节跳动火山引擎团队开源的强化学习（RL）训练框架，我们正迈向真正“能行动”的智能代理时代。

verl 的核心使命是为大型语言模型（LLMs）提供高效、可扩展的后训练能力。它基于 HybridFlow 论文实现，不仅支持标准的 RLHF 流程，更关键的是，它原生支持多工具调用、多轮对话交互和复杂任务编排。这意味着，我们可以训练出能够自主使用搜索、计算器、代码解释器等外部工具的 AI 智能体。

本文将聚焦一个极具代表性的实战场景：搜索 + 计算一体化。我们将展示如何利用 verl 构建一个能自动完成“信息检索 → 数据提取 → 数值计算 → 结果验证”全流程的智能代理系统。这不是简单的功能堆砌，而是通过强化学习让模型学会在正确的时间、以正确的方式调用正确的工具。

2. 核心架构：verl 的多工具协同机制

2.1 工具调用的标准化接口

verl 对所有外部工具采用统一的抽象接口设计，确保不同功能的工具可以无缝集成到同一个训练流程中。每个工具都继承自BaseTool类，并实现以下核心方法：

class BaseTool: async def create(self, instance_id: Optional[str] = None, **kwargs) -> tuple[str, ToolResponse]: """创建工具实例""" pass @rollout_trace_op async def execute(self, instance_id: str, parameters: dict[str, Any], **kwargs) -> tuple[ToolResponse, float, dict]: """执行工具操作""" pass async def calc_reward(self, instance_id: str, **kwargs) -> float: """计算工具奖励""" pass async def release(self, instance_id: str, **kwargs) -> None: """释放工具实例""" pass

这种设计使得无论是搜索 API、代码沙箱还是数学求解器，都能以一致的方式被调用和管理。

2.2 多轮对话与工具调度

verl 支持最大5轮的多轮对话训练，允许模型在一次任务中多次调用不同工具。整个流程由BaseInteraction类控制：

class BaseInteraction: async def start_interaction(self, instance_id: Optional[str] = None, **kwargs) -> str: """初始化交互""" async def generate_response(self, instance_id: str, messages: list[dict], **kwargs) -> tuple[bool, str, float, dict]: """生成响应并决定是否终止""" async def calculate_score(self) -> float: """计算最终得分""" async def finalize_interaction(self) -> None: """清理资源"""

在每一轮中，模型可以根据当前上下文决定是直接回复用户，还是调用某个工具来获取更多信息或执行计算。

2.3 工具注册与配置管理

所有可用工具通过 YAML 配置文件集中管理，便于灵活扩展和定制：

tools: - class_name: "verl.tools.search_tool.SearchTool" config: retrieval_service_url: "https://api.search-engine.com/v1/retrieve" num_workers: 100 rate_limit: 100 timeout: 30 topk: 3 tool_schema: name: "web_search" description: "Search the web for up-to-date information." parameters: type: "object" properties: query_list: type: "array" items: type: "string" description: "List of search queries" required: ["query_list"] - class_name: "verl.tools.sandbox_fusion_tools.SandboxFusionTool" config: sandbox_fusion_url: "https://api.sandbox.com/run_code" default_language: "python" memory_limit_mb: 1024 tool_schema: name: "code_interpreter" description: "Execute Python code in a secure environment." parameters: type: "object" properties: code: type: "string" description: "The code to execute." required: ["code"]

这样的配置方式让开发者无需修改核心代码即可添加新工具。

3. 实战案例：构建“搜索+计算”智能代理

3.1 场景设定：实时数据驱动的决策问题

假设我们要解决这样一个问题：

“截至2024年底，特斯拉在全球共生产了多少辆电动汽车？如果这些车全部在中国上路，相当于中国现有新能源汽车保有量的百分之多少？”

这个问题无法仅靠模型内部知识回答，因为它涉及两个动态数据点：

1. 特斯拉的历史累计产量（需搜索）
2. 中国新能源汽车保有量（需搜索）

然后还需要进行百分比计算。这就是典型的“搜索+计算”任务。

3.2 任务分解与执行流程

我们的智能代理需要按以下步骤完成任务：

1. 第一轮：发起第一次搜索

   • 调用web_search工具查询“特斯拉 历史累计电动汽车产量 截至2024年”
   • 获取返回结果中的关键数字

2. 第二轮：发起第二次搜索

   • 调用web_search工具查询“中国 新能源汽车 保有量 2024年”
   • 提取最新统计数据

3. 第三轮：执行数学计算

   • 将两个数值代入公式：(特斯拉产量 / 中国保有量) * 100
   • 调用code_interpreter工具运行计算代码

4. 第四轮：生成最终回答

   • 整合所有信息，形成自然语言回复

3.3 完整代码示例

以下是该任务的核心实现逻辑：

async def solve_real_world_question(): # 初始化工具 search_tool = SearchTool.from_config(search_config) code_tool = SandboxFusionTool.from_config(code_config) # 第一步：搜索特斯拉产量 search_result_1 = await search_tool.execute( instance_id="search_tesla", parameters={"query_list": ["特斯拉 2024年 累计电动车产量"]} ) tesla_output = extract_number_from_text(search_result_1.text) # 如：300万 # 第二步：搜索中国新能源车保有量 search_result_2 = await search_tool.execute( instance_id="search_china_ev", parameters={"query_list": ["中国 2024年 新能源汽车 保有量"]} ) china_total = extract_number_from_text(search_result_2.text) # 如：2000万辆 # 第三步：执行百分比计算 calculation_code = f""" # 计算占比 tesla_cars = {tesla_output} china_ev_total = {china_total} percentage = (tesla_cars / china_ev_total) * 100 print(f"{{percentage:.2f}}%") """ calc_result = await code_tool.execute( instance_id="calc_percentage", parameters={"code": calculation_code} ) final_percentage = parse_float_from_output(calc_result.text) # 第四步：生成最终回答 response = ( f"根据截至2024年底的数据，特斯拉累计生产了约{tesla_output}万辆电动汽车。\n" f"同期中国新能源汽车保有量约为{china_total}万辆。\n" f"因此，特斯拉产量约占中国总量的{final_percentage:.2f}%。" ) return response

3.4 在强化学习中训练此类行为

为了让模型学会这种多步推理和工具调用策略，我们需要在训练数据中注入类似的模式。例如，在 GSM8K 或 Search-R1 数据集上进行微调时，构造包含显式工具调用路径的样本：

{ "prompt": [ {"role": "user", "content": "北京到上海高铁最快需要多久？票价是多少？"} ], "extra_info": { "need_tools_kwargs": true, "tools_kwargs": { "web_search": { "create_kwargs": {} } }, "interaction_kwargs": { "allowed_tools": ["web_search", "code_interpreter"] } }, "response": "我需要查询最新的高铁时刻表和票价信息，请稍等。", "tool_calls": [ { "name": "web_search", "arguments": { "query_list": ["京沪高铁 最快时间 票价"] } } ] }

通过大量此类数据的训练，模型会逐渐掌握何时该搜索、何时该计算、以及如何组合多个工具来解决问题。

4. 性能优化与工程实践

4.1 并发控制与速率限制

在实际部署中，频繁调用外部API可能导致限流或延迟过高。verl 使用 Ray 分布式框架实现高效的并发管理，并采用令牌桶算法进行精确的速率控制：

@ray.remote(concurrency_groups={"acquire": 1, "release": 10}) class TokenBucketWorker: def __init__(self, rate_limit: int): self.rate_limit = rate_limit self._semaphore = threading.Semaphore(rate_limit) @ray.method(concurrency_group="acquire") def acquire(self): self._semaphore.acquire() @ray.method(concurrency_group="release") def release(self): self._semaphore.release()

每个工具实例在调用前必须先获取令牌，从而保证整体请求频率不超标。

4.2 连接池与缓存机制

为了提升性能，verl 还支持：

• HTTP连接池：复用底层TCP连接，减少握手开销
• 结果缓存：对高频查询（如常见数学常数、固定知识）进行本地缓存，避免重复请求

tools: - class_name: "verl.tools.search_tool.SearchTool" config: use_connection_pool: true pool_size: 20 enable_cache: true cache_ttl_seconds: 3600 # 缓存1小时

4.3 错误处理与重试策略

网络不稳定是常态。verl 内置了完善的错误恢复机制：

def call_search_api_with_retry(url, query_list, max_retries=5): for attempt in range(max_retries): try: response = requests.post(url, json={"queries": query_list}, timeout=10) if response.status_code == 200: return response.json() elif response.status_code in [500, 502, 503, 504]: time.sleep(2 ** attempt) # 指数退避 continue except (ConnectionError, Timeout) as e: time.sleep(2 ** attempt) continue raise RuntimeError("Search API failed after maximum retries")

这一机制确保即使在网络波动时，系统也能保持稳定运行。

5. 应用边界与未来展望

5.1 当前已验证的应用场景

verl 的多工具协同能力已在多个领域得到验证：

5.2 可扩展性设计

verl 的模块化架构使其极易扩展。你可以轻松接入新的工具类型，例如：

class WeatherAPITool(BaseTool): async def execute(self, instance_id: str, parameters: dict) -> tuple[ToolResponse, float, dict]: city = parameters.get("city") data = fetch_weather_data(city) return ToolResponse(text=str(data)), 1.0, {} # 注册到系统 register_tool("get_weather", WeatherAPITool)

只需几行代码，就能让模型具备查询天气的能力。

5.3 向通用智能代理演进

未来的方向是让模型不仅能被动执行预设工具链，更能自主规划任务路径。结合思维链（Chain-of-Thought）和强化学习，我们可以训练模型学会：

• 判断问题是否需要外部工具
• 规划最优的工具调用顺序
• 根据中间结果动态调整策略
• 自主发现并集成新工具

这正是 verl 所追求的目标：从“响应式AI”走向“行动式AI”。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。