注：本文包含 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_asyncexecute_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参数生效点：1
2
3
4
5
6
7
8
9
10
11
12
13
14配置 (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. 的推广形式）
- 这个数字在其他场景一般也不修改，因为分配时是比较的方式进行的，常数项影响有限
- 使用下面的方法 估算 FLOPs，用 Karmarkar-Karp 多路划分让各 micro-batch 工作量接近
注：长序列下强烈建议开启，需要 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
- Actor 的前向后向过程
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 完成实际生成
- 整体包含关系图
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26RayPPOTrainer (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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26# 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
LLMServerManager负责：- 启动 vLLM/SGLang replicas （Ray actor，占 GPU）
- 创建全局 load_balancer （Ray actor，做 least-inflight 负载均衡）
- 提供
get_client()(./verl/workers/rollout/llm_server.py)：返回一个轻量的LLMServerClient实例1
2
3
4
5
6def 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)
持有关系：
1
2
3
4
5class 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）：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16async 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
- 它们不直接持有 GPU，而是通过
AgentLoopWorker 的 调用入口（./verl/experimental/agent_loop/agent_loop.py）：
1
2
3
4
5
6
7
8async 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)
持有关系：
1
2
3
4
5
6
7class 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)：1
2
3
4
5
6
7
8
9
10
11async 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)：1
2
3
4
5
6
7
8
9
10
11
12
13
14async 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.run()为例 (./verl/experimental/agent_loop/single_turn_agent_loop.py)：1
2
3
4
5
6
7
8
9async 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
1
2
3
4
5
6
7
8
9
10class 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 outputserver是vLLMReplica(Ray actor, GPU) 的 handle，内部self.engine = AsyncLLM是进程内引擎- 理解：verl 不做任何 Batch 处理，都是为每个请求发送（Ray RPC 调度）生成命令到 LLMServer，LLMServer 会根据收到的请求做 Continuous Batching(即连续批处理)，在 LLMServer 这里实现并行调度（verl 的各个请求之间不知道对方的存在）
完整调用链路（一次 generate_sequences 调用）
调用流程详细梳理
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55RayPPOTrainer.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 actorLLMServerClient 的角色 轻量路由器，不持有 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.py1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18tools:
- 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.py1
2
3
4
5
6
7
8
9
10from verl.tools.function_tool import function_tool
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)给的约束）具体示例如下：
1
2
3
4
5
6
7
8
9
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)解析结果：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48{
"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 {}