NLP——VeRL源码阅读笔记

注：本文包含 AI 辅助创作

注：本文的版本包含一些很早的代码（部分小节可能有点过期了），也有很多是主要是截止到 20260530 日的 verl 版本 (commit_id=9f73954a87e247de4c31bc2f5222969e395b2904)

VeRL 中，RayWorkerGroup 函数调用路径理解

以 update_weights 完整调用链为例
- RayWorkerGroup 实例的 update_weights 方法在底层 Worker 类上被 @register(dispatch_mode=Dispatch.ONE_TO_ALL, blocking=False) 装饰
  - 这意味着 RayWorkerGroup 会通过 func_generator 动态生成一个包装函数
- 在 func_generator 中，调用流程为：
  - dispatch_fn → dispatch_one_to_all（透传参数）
  - execute_fn → execute_all → execute_all_async
  - execute_all_async 内部为每个 worker 调用 _execute_remote_single_worker，该方法返回 remote_call.remote(...)**，即 **ray.ObjectRef
  - 最终返回 [ray.ObjectRef, ray.ObjectRef, ...] — 一个 ray.ObjectRef 列表
  - 由于 blocking=False，ray.get() 不会在包装函数内执行
  - collect_fn → collect_all_to_all，直接透传返回（仍然是一个列表）
- 所以最终，RayWorkerGroup.update_weights 函数得到的结果是一个 ray.ObjectRef 列表

VeRL 中，`use_dynamic_bsz` 参数解读

动态 Batch Size 一般是指通过平衡每张卡上的 Attention Token（包括 Prompt 和 Response），防止出现现存爆炸而设计的策略
动态 Batch Size 这个策略只影响如何将 mini-batch 拆分为 micro-batch，不影响 train_batch_size 拆分为 mini-batch 的方式
从数学上看，这个策略本身并不修改梯度的期望（数学上使用前后等价），但是从实践上看，因为存在着一些浮点非结合律 + KK 重排样本等导致并不能 bit-wise 等价

actor.use_dynamic_bsz 参数生效点：

配置 (engine.yaml)
↓
ActorWorker.train_batch / infer_batch 注入 (engine_workers.py)
↓ tu.assign_non_tensor(data, use_dynamic_bsz=..., max_token_len_per_gpu=...)
engine.forward_backward_batch
↓
prepare_micro_batches (engine/utils.py)         ← 开关分叉点
├─ True:  rearrange_micro_batches (seqlen_balancing.py:348)
└─ False: tu.chunk_tensordict (固定 mbs 等分)
↓
for micro_batch in micro_batches:                     ← 内层循环
loss.backward()  (或 forward_only)
↓
postprocess_batch_func → restore_dynamic_batch        ← 还原顺序

一般做法：基于 KK 算法均衡划分：
- 使用下面的方法估算 FLOPs，用 Karmarkar-Karp 多路划分让各 micro-batch 工作量接近
  $$ \text{calculate_workload} \approx 24576 \cdot \text{seq_len} + \text{seq_len}^2 $$
- 24576 是根据标准的 7B dense transformer FLOPs 估算（Megatron-LM 论文 / Kaplan et al. 的推广形式）
- 这个数字在其他场景一般也不修改，因为分配时是比较的方式进行的，常数项影响有限
注：长序列下强烈建议开启，需要 bitwise 复现算法程序时注意这个参数是否都是打开或者关闭的
RL 中可以设置这个策略的地方有三个：
- Actor 的前向后向过程
  - actor.use_dynamic_bsz / actor.ppo_max_token_len_per_gpu
- 计算 $\pi_\text{old}$ 的 log_prob 计算（forward-only）
  - rollout.log_prob_use_dynamic_bsz / rollout.log_prob_max_token_len_per_gpu
- 计算 $\pi_\text{ref}$ 的 log_prob 计算（forward-only）
  - ref.log_prob_use_dynamic_bsz / ref.log_prob_max_token_len_per_gpu

RayPPOTrainer → AsyncRolloutManager → AgentLoopWorker 链路梳理

注：本节讲解代码 commit_id=9f73954a87e247de4c31bc2f5222969e395b2904
整条链路是”Driver 编排 → CPU Worker 并发协调 → GPU Replica 推理“的三层结构
Driver 的 RayPPOTrainer 持有三个 Manager（LLMServerManager 管 GPU 推理引擎、RewardLoopManager 管奖励计算、AgentLoopManager 管生成编排），三者通过 LLMServerClient（轻量路由器）和 reward_loop_worker_handles（Ray actor 引用）连接
- AgentLoopManager 持有 N 个纯 CPU 的 AgentLoopWorker（Ray actor），每个 Worker 在进程内用 asyncio 并发为每个样本实例化一个 AgentLoopBase 子类（SingleTurnAgentLoop/ToolAgentLoop）
- AgentLoopBase 子类通过 LLMServerClient → Ray RPC 调用 GPU 上的 vLLM replica 完成实际生成

整体包含关系图

RayPPOTrainer (Driver, 单进程)
│
├── owns: LLMServerManager                    ← 管理推理引擎生命周期
│   ├── owns: global_load_balancer (Ray actor) ← 全局负载均衡
│   ├── owns: rollout_replicas: list[RolloutReplica]  (Ray actor, GPU)
│   │   └── vLLMHttpServer / SGLangServer
│   │       └── self.engine = AsyncLLM (进程内引擎)
│   └── get_client() → LLMServerClient (轻量, 仅持有 load_balancer handle)
│
├── owns: RewardLoopManager                   ← 管理奖励计算
│   └── owns: reward_loop_workers: list[RewardLoopWorker] (Ray actor)
│
├── owns: AgentLoopManager (= self.async_rollout_manager)  ← 管理生成
│   ├── holds: llm_client: LLMServerClient     ← 从 LLMServerManager.get_client() 获得
│   ├── holds: reward_loop_worker_handles      ← 从 RewardLoopManager 获得(可选)
│   └── owns: agent_loop_workers: list[AgentLoopWorker] (Ray actor, CPU)
│       ├── holds: llm_client: LLMServerClient  ← 共享同一个 client 引用
│       ├── holds: reward_loop_worker_handles   ← 流式 reward 时使用
│       └── per-sample instantiate → AgentLoopBase (hydra.utils.instantiate)
│           ├── SingleTurnAgentLoop  (注册名 "single_turn_agent")
│           └── ToolAgentLoop        (注册名 "tool_agent")
│               └── holds: server_manager: LLMServerClient
│
└── owns: CheckpointEngineManager
    ├── holds: trainer = actor_rollout_wg
    └── holds: replicas = rollout_replicas

各类的职责与创建时机

RayPPOTrainer (Driver)

verl/trainer/ppo/ray_trainer.py

在 init_workers() 中按顺序创建链路上的所有组件：

# 1. 创建 WorkerGroup (Actor/Critic/Ref, 已在前面梳理过)
self.actor_rollout_wg = all_wg[str(actor_role)]

# 2. 创建 RewardLoopManager
self.reward_loop_manager = RewardLoopManager(config, rm_resource_pool)

# 3. 创建 LLMServerManager (持有 vLLM replicas)
self.llm_server_manager = LLMServerManager.create(
    config=self.config,
    worker_group=self.actor_rollout_wg,          # 用于 colocation 权重同步
    rollout_resource_pool=actor_rollout_resource_pool
)

# 4. 创建 AgentLoopManager (= async_rollout_manager)
self.async_rollout_manager = AgentLoopManager.create(
    config=self.config,
    llm_client=self.llm_server_manager.get_client(),      # ← 关键: 注入 LLMServerClient
    teacher_client=...,
    reward_loop_worker_handles=reward_loop_worker_handles  # ← 可选: 注入 RewardLoopWorker
)

# 5. 创建 CheckpointEngineManager
self.checkpoint_manager = CheckpointEngineManager(
    trainer=self.actor_rollout_wg,
    replicas=self.llm_server_manager.get_replicas(),
)

LLMServerManager → LLMServerClient

verl/workers/rollout/llm_server.py

LLMServerManager 负责：

启动 vLLM/SGLang replicas （Ray actor，占 GPU）
创建全局 load_balancer （Ray actor，做 least-inflight 负载均衡）

提供 get_client() (./verl/workers/rollout/llm_server.py)：返回一个轻量的 LLMServerClient 实例

def get_client(self, client_cls=LLMServerClient, **kwargs) -> LLMServerClient:
    return client_cls(
        config=self.config,
        load_balancer_handle=self.global_load_balancer,
        **kwargs,
    )

注意：LLMServerClient 本身不持有任何 GPU 资源 ，它只是一个”路由器”，持有 load_balancer 的 Ray actor handle
- 多个 AgentLoopWorker 共享同一个 LLMServerClient 实例（或各自的副本），它们最终都路由到同一组 vLLM replicas

AgentLoopManager (= async_rollout_manager)

verl/experimental/agent_loop/agent_loop.py

持有关系：

class AgentLoopManager:
    def __init__(self, config, llm_client, teacher_client, reward_loop_worker_handles):
        self.llm_client = llm_client                    # LLMServerClient 实例
        self.reward_loop_worker_handles = reward_loop_worker_handles  # list[ActorHandle] 或 None
        self.agent_loop_workers_class = ray.remote(AgentLoopWorker)

创建 Worker（./verl/experimental/agent_loop/agent_loop.py）：

async def _init_agent_loop_workers(self):
    self.agent_loop_workers = []
    num_workers = self.rollout_config.agent.num_workers  # 默认 8, 纯 CPU
    for i in range(num_workers):
        node_id = node_ids[i % len(node_ids)]  # round-robin 跨节点
        self.agent_loop_workers.append(
            self.agent_loop_workers_class.options(
                name=f"agent_loop_worker_{i}_...",
                scheduling_strategy=NodeAffinitySchedulingStrategy(node_id, soft=True)
            ).remote(
                self.config,
                self.llm_client,              # ← 传给每个 worker
                self.teacher_client,
                self.reward_loop_worker_handles,  # ← 传给每个 worker
            )
        )

AgentLoopWorker 是纯 CPU 的 Ray actor，分布在集群各节点上
- 它们不直接持有 GPU，而是通过 llm_client 远程调用 GPU 上的 vLLM replicas

AgentLoopWorker 的调用入口（./verl/experimental/agent_loop/agent_loop.py）：

async def generate_sequences(self, prompts: DataProto) -> DataProto:
    chunkes = prompts.chunk(len(self.agent_loop_workers))  # 按 worker 数切分
    outputs = await asyncio.gather(
        *[worker.generate_sequences.remote(chunk)           # 并行 Ray RPC
        for worker, chunk in zip(self.agent_loop_workers, chunkes)]
    )
    output = DataProto.concat(outputs)                      # 合并结果
    return output

AgentLoopWorker (Ray actor, CPU)

verl/experimental/agent_loop/agent_loop.py

持有关系：

class AgentLoopWorker:
    def __init__(self, config, llm_client, teacher_client, reward_loop_worker_handles):
        self.llm_client = llm_client          # LLMServerClient (用于调 vLLM)
        self.reward_loop_worker_handles = reward_loop_worker_handles  # 流式 reward
        self.tools = load_all_tools(...)      # 工具列表
        self.tokenizer = ...
        self.processor = ...

核心方法 generate_sequences (./verl/experimental/agent_loop/agent_loop.py)：

async def generate_sequences(self, batch: DataProto) -> DataProto:
    sampling_params = dict(temperature=..., top_p=..., top_k=..., ...)
    
    # 对 batch 中每个样本, 创建一个 asyncio Task
    tasks = []
    for i in range(len(batch)):
        tasks.append(asyncio.create_task(
            self._run_agent_loop(sample_sampling_params, trajectory_info[i], **kwargs)
        ))
    outputs = await asyncio.gather(*tasks)  # 并发执行所有样本的 agent loop
    return self._postprocess(outputs)

一个 AgentLoopWorker 处理一个 chunk（多个样本），每个样本有一个 agent loop，所有样本的 agent loop 都在 同一个 Ray actor 进程内并发执行（asyncio），而不是每样本一个 actor
- 理解：一个 Worker 一个进程，对应一个 Ray actor，这个 Worker 负责的所有样本都并发进行（通过协程的方式）

_run_agent_loop 的实现 (./verl/experimental/agent_loop/agent_loop.py)：

async def _run_agent_loop(self, sampling_params, trajectory, *, agent_name, **kwargs):
    agent_loop_config = _agent_loop_registry[agent_name]  # 从注册表取配置（包含 _target_ 字段）
    agent_loop = hydra.utils.instantiate(                 # 每次调用都新实例化一个 AgentLoop
        config=agent_loop_config,
        trainer_config=DictConfigWrap(config=self.config),
        server_manager=self.llm_client,    # ← 注入 LLMServerClient
        tokenizer=self.tokenizer,
        processor=self.processor,
        dataset_cls=self.dataset_cls,
        data_config=...,
        tools=ToolListWrap(self.tools),
    )
    output = await agent_loop.run(sampling_params, **kwargs)  # 执行 agent loop
    return await self._agent_loop_postprocess(output, ...)

AgentLoopBase (每样本实例化)

verl/experimental/agent_loop/agent_loop.py
抽象基类，每个样本创建一个实例（ _run_agent_loop 函数中通过 hydra.utils.instantiate），持有 server_manager: LLMServerClient

子类注册名文件用途

SingleTurnAgentLoop "single_turn_agent" single_turn_agent_loop.py 单轮生成（默认）

ToolAgentLoop "tool_agent" tool_agent_loop.py 多轮工具调用

子类	注册名	文件	用途
`SingleTurnAgentLoop`	`"single_turn_agent"`	single_turn_agent_loop.py	单轮生成（默认）
`ToolAgentLoop`	`"tool_agent"`	tool_agent_loop.py	多轮工具调用

以 SingleTurnAgentLoop.run() 为例 (./verl/experimental/agent_loop/single_turn_agent_loop.py)：

async def run(self, sampling_params, **kwargs) -> AgentLoopOutput:
    prompt_ids = await self.apply_chat_template(messages, ...)
    output: TokenOutput = await self.server_manager.generate(  # ← 调 LLMServerClient
        request_id=...,
        prompt_ids=prompt_ids,
        sampling_params=sampling_params,
        ...
    )
    return AgentLoopOutput(prompt_ids=..., response_ids=..., ...)

LLMServerClient → vLLM Replica

verl/workers/rollout/llm_server.py

class LLMServerClient:
    async def generate(self, request_id, *, prompt_ids, sampling_params, ...):
        server_id, server = await self._acquire_server(request_id)  # load_balancer 选 replica
        output = await server.generate.remote(                      # Ray RPC → vLLM replica
            request_id=uuid4().hex,
            prompt_ids=prompt_ids,
            sampling_params=sampling_params,
            ...
        )
        return output

server 是 vLLMReplica (Ray actor, GPU) 的 handle，内部 self.engine = AsyncLLM 是进程内引擎
理解：verl 不做任何 Batch 处理，都是为每个请求发送（Ray RPC 调度）生成命令到 LLMServer，LLMServer 会根据收到的请求做 Continuous Batching(即连续批处理)，在 LLMServer 这里实现并行调度（verl 的各个请求之间不知道对方的存在）

完整调用链路（一次 `generate_sequences` 调用）

调用流程详细梳理

RayPPOTrainer.fit()
│  (Driver 进程, Python 调用)
│
├─ self.async_rollout_manager.generate_sequences(combined_gen_batch)
│     │
│     │  [AgentLoopManager, Driver 进程内]
│     ├─ prompts.chunk(N)  → 切成 N 份
│     ├─ asyncio.gather(*[
│     │     worker.generate_sequences.remote(chunk)  ──────┐
│     │  ])                                                 │ Ray RPC
│     │                                                     ▼
│     │                          [AgentLoopWorker #i, CPU Ray actor]
│     │                          ├─ for each sample in chunk:
│     │                          │    asyncio.create_task(
│     │                          │        self._run_agent_loop(...)
│     │                          │    )
│     │                          ├─ asyncio.gather(*tasks)
│     │                          │    │
│     │                          │    │  [每样本, 同进程内]
│     │                          │    ├─ hydra.utils.instantiate → AgentLoopBase 子类
│     │                          │    │    (SingleTurnAgentLoop / ToolAgentLoop)
│     │                          │    │
│     │                          │    ├─ agent_loop.run(sampling_params)
│     │                          │    │    │
│     │                          │    │    │  [AgentLoopBase, 同进程内]
│     │                          │    │    ├─ apply_chat_template → prompt_ids
│     │                          │    │    ├─ self.server_manager.generate(prompt_ids, ...)
│     │                          │    │    │    │
│     │                          │    │    │    │  [LLMServerClient, 同进程内]
│     │                          │    │    │    ├─ _acquire_server(request_id)
│     │                          │    │    │    │    → load_balancer.acquire_server.remote()
│     │                          │    │    │    │    → 返回 (server_id, vLLMReplica handle)
│     │                          │    │    │    ├─ server.generate.remote(prompt_ids, ...)
│     │                          │    │    │    │    │
│     │                          │    │    │    │    │ Ray RPC
│     │                          │    │    │    │    ▼
│     │                          │    │    │    │  [vLLMHttpServer, GPU Ray actor]
│     │                          │    │    │    │  ├─ self.engine.generate(prompt, sampling_params)
│     │                          │    │    │    │  │    │ engine.generate 被并发调用时，LLM Engine 会自动实现 Continuous Batching(并发推理)
│     │                          │    │    │    │  │    │ 进程内 Python 调用
│     │                          │    │    │    │  │    ▼
│     │                          │    │    │    │  │  [AsyncLLM engine]
│     │                          │    │    │    │  │  └─ 采样生成 token → RequestOutput
│     │                          │    │    │    │  │
│     │                          │    │    │    │  └─ return TokenOutput
│     │                          │    │    │    └─ return TokenOutput
│     │                          │    │    └─ return AgentLoopOutput
│     │                          │    └─ return _InternalAgentLoopOutput
│     │                          ├─ _postprocess → DataProto
│     │                          └─ return DataProto
│     │                                                     │
│     ├─ DataProto.concat(outputs)  ◄───────────────────────┘
│     └─ return DataProto
│
└─ 继续 PPO 流程 (reward → old_log_prob → ref → value → advantage → update)

注：engine.generate 被并发调用时，LLM Engine 会自动实现 Continuous Batching(并发推理)，这完全由引擎消化，所以 verl 框架不需要自己再实现 Batch 处理了，只需要一个个把请求并发分发给 LLM Engine 即可（vLLM 和 sglang 均能实现）

补充：其他关键设计总结

关键设计总结表

维度	设计
Manager 所在进程	`AgentLoopManager`、`LLMServerManager`、`RewardLoopManager` 都在 Driver 进程内，是普通 Python 对象
Worker 所在进程	`AgentLoopWorker`（CPU Ray actor）、`vLLMReplica`（GPU Ray actor）、`RewardLoopWorker`（CPU/GPU Ray actor）都是独立 Ray actor
LLMServerClient 的角色	轻量路由器，不持有 GPU，被 `AgentLoopWorker` 和 `AgentLoopBase` 共享持有
AgentLoopBase 的生命周期	每样本实例化（`hydra.utils.instantiate`），用完即弃；`_run_agent_loop` 每次调用都 new 一个
并发模型	两层并发：① Manager 层：N 个 Worker 跨 actor 并行（Ray）；② Worker 层：chunk 内多个样本 asyncio 并发（单进程协程）
CPU/GPU 分离	`AgentLoopWorker` 纯 CPU（编排 + tokenizer + tool 执行），vLLM replica 在 GPU 上；两者通过 Ray RPC 解耦
reward_loop_worker_handles 的传递	可选注入；启用 streaming reward 时，`AgentLoopWorker` 在 agent loop 内直接调用 reward worker，实现生成与奖励计算的 Pipeline 并行
权重同步	`CheckpointEngineManager` 持有 `actor_rollout_wg`（训练侧）和 `replicas`（推理侧），训练更新后通过 IPC 把权重推到 vLLM engine

VeRL NativeTool 和 FunctionTool 比较

NativeTool 与 FunctionTool 的定位差异

位置：两者都通过 verl/tools/tool_registry.py 的 load_all_tools 函数加载和合并返回
核心：
- 工具需要 per-trajectory 状态、异步生命周期、自定义 step reward，或需要外部资源（沙盒、连接池）-> NativeTool
- 工具是无状态纯函数，schema 能从签名/docstring 推断 -> FunctionTool ，配置成本显著更低

定位不同：

维度	NativeTool（`BaseTool` 子类）	FunctionTool（`@dataclass`）
状态模型	有状态，per-trajectory `instance_id`，提供 `create`/`execute`/`calc_reward`/`release` 全生命周期	无状态，单次 `call(parameters)`
Schema 来源	YAML 显式声明，或由 `get_openai_tool_schema()` 方法生成	由函数签名 + Google-style docstring 自动推断（`transformers.utils.get_json_schema`）
配置形态	YAML 文件，经 `tool_config_path` 指定	Python 文件，经 `function_tool_path` 指定，用 `@function_tool` 装饰器注册
加载缓存	每次 `load_all_tools` 都重新实例化（两次初始化的对象必须不同测试代码中有 ``）	全局 `FUNCTION_TOOL_REGISTRY` 缓存，同一 path 只 import 一次
执行入口	`async execute(instance_id, parameters) -> (ToolResponse, reward, metrics)`	`async call(parameters) -> Any`，再经 `normalize_function_tool_return` 归一化
同步/异步	`execute` 恒为协程	装饰器自动检测 `iscoroutinefunction`，同步函数用 `asyncio.to_thread` 包装
奖励信号	内置 `calc_reward` 钩子，可基于 tool state 产出 step reward	仅靠返回值 `(response, reward, metrics)` 元组携带，默认 0.0
典型场景	沙盒执行、搜索引擎、爬虫等需要会话状态/外部资源管理的复杂工具	计算器、天气查询等无状态的工具函数

配置上二者 可共存
- 这两个工具调用的配置默认见 verl/trainer/config/rollout/rollout.yaml
  - 两个字段是可以同时配置的
- 注意：工具名必须全局唯一
  - 对于重复的情况： tools/tool_registry.py 中的 load_all_tools 函数会检查会抛 ValueError

NativeTool 与 FunctionTool 典型配置

NativeTool，YAML 配置（`tool_config_path`）

来自下面 tests/tools/test_mixed_tools_on_cpu.py ，test_mixed_tools_on_cpu.py

tools:
    - class_name: "tests.tools._stub_search_tools.StubSearchTool"  # BaseTool 子类全限定名
    config:
        retrieval_service_url: http://stub/retrieve   # 业务参数
        topk: 3
        type: native                                  # ToolType.NATIVE
    tool_schema:                                     # OpenAI function schema 显式声明
        type: function
        function:
            name: search
            description: Stub web search.
            parameters:
                type: object
                properties:
                    query_list:
                        type: array
                        description: A list of fully-formed semantic queries.
                required: ["query_list"]

对应的 BaseTool 子类实现见 examples/tutorial/agent_loop_get_started/sandbox.py，需重写 execute（必要时还有 create/release/calc_reward）

FunctionTool，Python 文件（`function_tool_path`）

来自 tests/experimental/agent_loop/function_tool_examples.py

from verl.tools.function_tool import function_tool

@function_tool               # name 默认取函数名；也可 @function_tool("custom_name")
def get_weather(city: str) -> dict:
    """Get the current weather for a city.

    Args:
        city: The city to look up, e.g. "Tokyo" or "San Francisco".
    """
    return {"temperature_c": 17.3, "condition": "drizzle"}

@function_tool 装饰器的注意事项：

@function_tool（verl/tools/function_tool.py）装饰器如果没有传入 custom_name 作为参数，则装饰内部会通过 _build_schema_from_fn 调 get_json_schema(fn) 自动推断 schema

自动推断 schema 时的函数需要满足以下条件：

必须带 Google-style docstring + 参数类型注解（注：这个要求来自最底层的 transformers.utils.get_json_schema(fn) 给的约束）

具体示例如下：

@function_tool
def get_weather(city: str) -> dict:             # 参数类型注解
    """Get the current weather for a city.       # description

    # 下面是 Google-style 参数定义：
    Args:
        city: The city to look up, e.g. "Tokyo".
    """
    ...

get_json_schema(get_weather) 解析结果：

                {
                    "type": "function",
                    "function": {
                        "name": "get_weather",
                        "description": "Get the current weather for a city.       # description\n\n# 下面是 Google-style 参数定义：",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "city": {
                                    "type": "string",
                                    "description": "The city to look up, e.g. \"Tokyo\"."
                                }
                            },
                            "required": [
                                "city"
                            ]
                        },
                        "return": {
                            "type": "object"
                        }
                    }
                }
                ``` 
* `FunctionTool.call` 的返回结果总是会被 `verl/tools/function_tool.py` 中的 `normalize_function_tool_return` 函数 统一为 `(ToolResponse, reward, metrics)`
    * `FunctionTool.call` （即 `fn`）的返回结果可以是 ToolResponse, str, dict, tuple 或其他任意可以转换成 str 的类型
        * 其实任何对象理论上都可以转换成 str，所以返回值可以是任意类型
        * 如果是 tuple 时，必须返回值在 1-3 个之间，不能是其他数字，对应 `response, reward, metrics`（一定会依次转换成 `(ToolResponse, float, dic)` 类型）
    * `verl/tools/function_tool.py` 中的 `normalize_function_tool_return` 函数定义
    ```python
    def normalize_function_tool_return(ret: Any) -> tuple[ToolResponse, float, dict]:
        if isinstance(ret, ToolResponse): # 如果结果已经是 ToolResponse，直接返回 ToolResponse, reward 0.0, metrics {}
            return ret, 0.0, {}
        if isinstance(ret, str): # 如果结果是字符串，将字符串转换为 ToolResponse, reward 0.0, metrics {}
            return ToolResponse(text=ret), 0.0, {}
        if isinstance(ret, dict): # 如果结果是字典，将字典转换为 ToolResponse, reward 0.0, metrics {}
            return ToolResponse(text=json.dumps(ret, ensure_ascii=False)), 0.0, {}
        if isinstance(ret, tuple): # 如果结果是元组，根据元组长度判断是否包含 reward 和 metrics
            if not 1 <= len(ret) <= 3:
                raise TypeError(
                    f"@function_tool return tuple must have length 1, 2, or 3 "
                    f"(got length {len(ret)}: {ret!r}). Use (response,), "
                    f"(response, reward), or (response, reward, metrics)."
                )
            response = _coerce_response(ret[0])
            reward = 0.0 if len(ret) < 2 or ret[1] is None else float(ret[1]) # 转换成 float 类型
            metrics = {} if len(ret) < 3 or ret[2] is None else dict(ret[2]) # 转换成字典
            return response, reward, metrics
        return ToolResponse(text=str(ret)), 0.0, {} # 将其他类型的结果转换为 ToolResponse, reward 0.0, metrics {}