Ray Serve DeploymentHandle 的异步批处理

Ray Serve 里，异步批处理要把两个概念接起来：调用方通过 DeploymentHandle.remote() 一次提交一个对象，部署内部用 @serve.batch 把多个对象聚合成列表。remote() 返回的是可等待的 DeploymentResponse，因此可以先把请求全部发出，再统一等待结果；真正的批处理函数则必须是异步函数，并返回与输入等长的结果列表。

官方地址：https://docs.ray.io/en/latest/serve/

下面的例子只讨论 Ray Serve 的 DeploymentHandle 和动态请求批处理，不把 HTTP 网关、模型版本路由或其他推理框架混进来。

先分清句柄调用和批处理函数

句柄负责把单次调用送到某个部署，批处理装饰器负责在部署内部收集这些单项输入。调用方看到的是一个个响应对象，批处理方法看到的则是一个列表。两层不要互换：不要让调用方手工拼一个“大列表”再把它当成每次请求的唯一参数，也不要让批处理方法只返回一个聚合值。

用 async def 接住一批输入

@serve.batch 装饰的方法要接收一个列表，并返回同样长度的列表。下面把每个输入乘以一个固定系数，故意保持逻辑简单，重点是展示批处理的边界：

from typing import List

from ray import serve


@serve.deployment
class ScoreService:
    @serve.batch(max_batch_size=8, batch_wait_timeout_s=0.02)
    async def score_batch(self, values: List[float]) -> List[float]:
        # 每个输入对应一个输出，不能把一批结果压成单个数字。
        return [value * 1.2 for value in values]

    async def __call__(self, value: float) -> float:
        # 单项调用交给批处理方法，Serve 负责在内部聚合。
        return await self.score_batch(value)


app = ScoreService.bind()

这里的 __call__ 仍然以单个 value 为入口，score_batch 才是批量函数。批大小达到 max_batch_size，或者首个请求等待时间达到 batch_wait_timeout_s，Serve 就会把当前积累的输入交给批处理函数。

先提交多个 remote，再等待响应

从另一个部署调用时，关键动作是不要在提交循环里立刻等待每个结果。每次 remote() 都会返回一个 DeploymentResponse；把这些响应先收集起来，调用已经排入 Serve 的路由与批处理队列，再按原顺序取回结果：

from typing import List

from ray import serve
from ray.serve.handle import DeploymentHandle


@serve.deployment
class BatchCaller:
    def __init__(self, scorer: DeploymentHandle):
        self.scorer = scorer

    async def __call__(self, values: List[float]) -> List[float]:
        # 先发出全部单项请求，让 Serve 有机会把它们放进同一批。
        responses = [self.scorer.remote(value) for value in values]

        # 逐个 await 仍保持结果顺序；请求已经在上一行全部提交。
        results = [await response for response in responses]
        return results


app = BatchCaller.bind(ScoreService.bind())

这段写法的重点不是把 DeploymentResponse 当作普通 asyncio.Future 改造，而是利用它本身的可等待语义。调用方可以按输入顺序读取结果；批处理是否凑满、由哪个副本处理，则由 Serve 的路由和批处理机制决定。

四个参数怎么和负载约束对齐

max_batch_size 是一次交给底层函数的最大批量度量，默认按列表长度计算；batch_wait_timeout_s 是首个请求到达后等待凑批的最长时间。前者偏大通常有利于吞吐但会增加内存压力和首批等待，后者偏大可能提高低流量下的批利用率，却会抬高延迟。

max_concurrent_batches 控制一个副本同时执行的批次数。只有底层异步计算确实能并发、并且资源允许时才适合增大它。若每个输入的成本差异很大，可以提供 batch_size_fn，让 Serve 按总 token 数、图节点数等自定义度量计算批大小，而不是只数请求个数。

from typing import List

from ray import serve


def token_count(texts: List[str]) -> int:
    # 用输入 token 总量作为批大小度量，而不是字符串条数。
    return sum(len(text.split()) for text in texts)


@serve.deployment
class TokenService:
    @serve.batch(
        max_batch_size=512,
        batch_wait_timeout_s=0.05,
        max_concurrent_batches=2,
        batch_size_fn=token_count,
    )
    async def infer_batch(self, texts: List[str]) -> List[int]:
        # 返回值仍必须与 texts 一一对应，长度不能变化。
        return [len(text.split()) for text in texts]

自定义度量时，max_batch_size 的单位也随之改变：示例中它表达 token 总量，不是最多 512 条文本。参数命名仍然是批大小，但判断条件已经由 batch_size_fn 定义。

用结果长度和指标判断是否正确

最直接的结果判断是：输入多少项，就返回多少项，并且响应顺序与提交顺序一致。若批处理函数漏掉一项、合并了多项或返回一个标量，Serve 会把它视为批量结果不匹配，而调用方也无法把响应准确对应回原请求。

参数调优不要只盯着总吞吐。Ray Serve 的批处理指标包括批等待时间、批执行时间、批队列长度和批利用率。等待时间持续偏高，说明等待窗口或队列压力可能过大；利用率偏低，可能是流量不足或等待窗口设置不合适；执行时间偏高，则要回到模型计算和副本资源本身。

落地时可以按这个顺序判断：

• 先确认每个 remote() 调用只提交一个业务对象，并保存每个响应。
• 再确认批处理函数是 async def，参数是列表，返回列表长度不变。
• 最后结合延迟目标选择等待窗口，并用批等待、队列长度和利用率观察实际效果。

常见问题

DeploymentHandle.remote() 会直接返回模型结果吗？

不会。它返回一个代表进行中调用的 DeploymentResponse；在异步代码中需要等待它，才能得到部署方法的实际结果。

为什么 max_batch_size 设为 8，却没有每次都得到 8 条输入？

批处理会在达到上限或等待窗口到期时发送当前批次。低流量下通常会先到达超时条件，因此实际批量可以小于上限；使用 batch_size_fn 时，判断依据还会从条数变成自定义度量。

批处理函数能返回一个总结果吗？

不适合这样做。动态请求批处理要求输出与输入逐项对应，应该返回与输入列表等长的结果列表，再由调用方按响应顺序取回。

归纳起来，DeploymentHandle 负责异步提交和响应句柄，@serve.batch 负责把单项请求聚成批次。先提交、后等待，配合等长返回和批处理指标，就能把吞吐、延迟与资源边界放在同一个可解释的模型里。