`limited.py` — 这个文件实现了带速率限制的奖励管理器 `RateLimitedRewardManager`¶

文件路径: verl/experimental/reward_loop/reward_manager/limited.py

文件概述¶

这个文件实现了带速率限制的奖励管理器 RateLimitedRewardManager，以及配套的异步令牌桶 AsyncTokenBucket。当奖励计算依赖外部 API（如 OpenAI API）时，需要控制请求频率以避免被限流。

一、AsyncTokenBucket - 异步令牌桶¶

令牌桶是经典的限流算法。这里的实现支持异步操作：

class AsyncTokenBucket:
    """异步令牌桶限流器"""

    def __init__(self, rate: float, capacity: float):
        self.rate = rate          # 令牌填充速率（每秒）
        self.capacity = capacity  # 桶的最大容量
        self.tokens = capacity    # 当前令牌数
        self.last_time = time.monotonic()

    async def acquire(self, tokens: int = 1):
        """获取指定数量的令牌，如果不够则等待"""
        while True:
            self._refill()  # 先补充令牌
            if self.tokens >= tokens:
                self.tokens -= tokens
                return
            # 令牌不够，等待直到有足够的令牌
            wait_time = (tokens - self.tokens) / self.rate
            await asyncio.sleep(wait_time)

工作原理： 1. 桶里有固定容量的"令牌" 2. 令牌以固定速率补充 3. 每次发送请求需要消耗令牌 4. 令牌不够时，异步等待直到有足够的令牌

二、RateLimitedRewardManager - 带限流的奖励管理器¶

三层速率限制¶

这个管理器实现了三层限流保护：

@register("rate_limited")
class RateLimitedRewardManager(RewardManagerBase):
    def __init__(self, config, tokenizer, ...):
        super().__init__(config, tokenizer, ...)

        # 第1层：并发数限制
        self.concurrency_limiter = asyncio.Semaphore(
            config.get("max_concurrency", 100)
        )

        # 第2层：每分钟请求数限制（RPM）
        self.rpm_bucket = AsyncTokenBucket(
            rate=config.get("rpm", 1000) / 60,
            capacity=config.get("rpm", 1000)
        )

        # 第3层：每分钟 Token 数限制（TPM）
        self.tpm_bucket = AsyncTokenBucket(
            rate=config.get("tpm", 100000) / 60,
            capacity=config.get("tpm", 100000)
        )

请求处理流程¶

async def _rate_limited_call(self, prompt, response, ...):
    """经过三层限流保护的单次请求"""
    # 1. 并发控制
    async with self.concurrency_limiter:
        # 2. RPM 限制
        await self.rpm_bucket.acquire(1)
        # 3. TPM 限制（根据 token 数量）
        token_count = len(self.tokenizer.encode(prompt + response))
        await self.tpm_bucket.acquire(token_count)

        # 实际调用评分函数
        score = await compute_score_fn(prompt, response)
        return score

兼容性支持¶

为了兼容旧版代码，这个类还实现了 __call__ 方法：

def __call__(self, data, ...):
    """兼容旧版接口，让管理器可以像函数一样被调用"""
    return self.run_single(data, ...)

核心类/函数列表¶

名称	类型	说明
`AsyncTokenBucket`	类	异步令牌桶限流器
`AsyncTokenBucket.acquire`	方法	获取令牌（不够则等待）
`RateLimitedRewardManager`	类	带三层限流的奖励管理器
`concurrency_limiter`	属性	最大并发数限制（Semaphore）
`rpm_bucket`	属性	每分钟请求数限制
`tpm_bucket`	属性	每分钟 Token 数限制

与其他模块的关系¶

继承自 RewardManagerBase
通过 @register("rate_limited") 注册到全局注册表
适用于依赖外部 API（如 OpenAI、Anthropic）的奖励计算场景

小结¶

RateLimitedRewardManager 解决了一个实际问题：当使用外部 API 计算奖励时，如何避免触发速率限制。它通过三层保护（并发数、RPM、TPM）确保请求不会超过 API 提供商的限制。这在使用 GPT-4 等模型做奖励评分时特别有用。

limited.py — 这个文件实现了带速率限制的奖励管理器 RateLimitedRewardManager¶