用 Jina 的 Executor 与 Deployment 部署一个基于 Stable Diffusion 的 gRPC AI 服务-平芜编程栈

后端
微服务
RPC框架
模型推理服务
人工智能

【免费下载链接】jina

☁️ Build multimodal AI applications with cloud-native stack

项目地址：https://gitcode.com/gh_mirrors/ji/jina

本篇技术指南以 Jina（jina-serve）开源框架为基准，讲解如何把任意 AI 模型封装成标准化的Executor微服务，再通过Deployment进行部署、暴露 gRPC 端点，并用Client完成端到端请求。全文以"文本生成图像"的 Stable Diffusion 服务为实战载体，覆盖 Executor 编写规范、@requests端点机制、Python/YAML 两种部署方式、GPU 与副本扩缩容以及动态批处理（dynamic batching）配置，读完后你将掌握一套可复用到 OCR、向量编码、PDF 表格抽取等任意模型服务的 Jina 部署范式。

开始之前，建议先阅读官方前置指南 docs/tutorials/before-you-start.md，其中涵盖 DocArray 数据模型、Jina 安装等预备知识。本教程写作时基于 Jina 3.14，理论上兼容后续版本。

图：一个 Deployment 服务一个 Executor，是 Jina 中最小的可部署服务单元

理解核心概念：Executor 与 Deployment

在 Jina-serve 中构建任何模型服务，始终围绕两个核心抽象：

Document 与 DocList：进出 Jina-serve 的所有数据都以 DocArray 包中的 Document（文档）和 DocList（文档列表）形式承载。前者定义单个数据单元（一段文本、一张图片或一个复杂多模态对象），后者是这些单元的数组集合。
Executor：一个自包含的 gRPC 微服务，对传入的 DocList 执行特定任务。任务可以简单到"把 Document 的文本全部大写"，也可以复杂到"为内容生成向量表征"或"根据文本提示生成图像"。
Deployment：负责把 Executor 跑起来、用副本（replicas）和分片（shards）进行水平扩展，并对外暴露可供客户端收发请求的服务端点。

从源码看，Deployment位于 jina/orchestrate/deployments/init.py，其类定义为"一个不可变的 Pod 集合，以副本形式运行，共享相同的输入与输出 socket"（见 jina/orchestrate/deployments/init.py#L142-L147），这正解释了 Deployment 是 Jina 中最小的、可独立对外服务的部署单元。

一个 Deployment 只服务一个 Executor。如果需要串联多个 Executor 组成流水线，请参考 docs/concepts/orchestration/flow.md 中的 Flow 教程。

安装前置依赖

本例需要安装两部分依赖：

Jina 框架本身（提供 Executor / Deployment / Client 等核心能力）；
待服务模型的具体依赖（本例为 Hugging Face 的 diffusers 库及 PyTorch）。

pip install jina pip install diffusers

运行StableDiffusionPipeline还需要 torch 与 transformers 等 diffusers 的传递依赖，请确保你的环境满足模型运行要求（如需 GPU 推理，请预先安装 CUDA 版 PyTorch）。

编写 Executor：实现模型服务逻辑

把服务逻辑写入text_to_image.py：

import numpy as np from jina import Executor, requests from docarray import BaseDoc, DocList from docarray.documents import ImageDoc class ImagePrompt(BaseDoc): text: str class TextToImage(Executor): def __init__(self, **kwargs): super().__init__(**kwargs) from diffusers import StableDiffusionPipeline import torch self.pipe = StableDiffusionPipeline.from_pretrained( "CompVis/stable-diffusion-v1-4", torch_dtype=torch.float16 ).to("cuda") @requests def generate_image(self, docs: DocList[ImagePrompt], **kwargs) -> DocList[ImageDoc]: images = self.pipe(docs.text).images # image here is in PIL format for i, doc in enumerate(docs): doc.tensor = np.array(images[i])

下面逐段拆解这段代码的四个组成部分。

导入语句

from docarray import DocList, BaseDoc

Document 和 DocList 来自 DocArray 包，是 Jina-serve 的原生 IO 格式——所有进出 Executor 的数据都必须以它们为载体。

from jina import Executor, requests

这是 Jina-serve 的 Executor 基类与requests装饰器，二者的配合构成了 Executor 端点机制的核心。

import numpy as np

NumPy 是本 Executor 特有的依赖，用于把 PIL 图像转换为tensor数值数组，并非 Jina 框架的通用要求。

定义文档类型

from docarray import BaseDoc from docarray.documents import ImageDoc class ImagePrompt(BaseDoc): text: str

Executor 的输入是自定义的ImagePrompt（仅含一个text字段），输出则是 DocArray 内置的ImageDoc（支持tensor、url等图像字段）。通过类型标注，Jina 会在运行时对请求数据做 schema 校验与序列化，这也让 Executor 的输入输出契约清晰可读。

Executor 类与构造方法

class TextToImage(Executor): def __init__(self, **kwargs): super().__init__(**kwargs) import torch from diffusers import StableDiffusionPipeline self.pipe = StableDiffusionPipeline.from_pretrained( "CompVis/stable-diffusion-v1-4", torch_dtype=torch.float16 ).to("cuda")

所有 Executor 都必须继承 Jina 的Executor基类（定义于 jina/serve/executors/init.py）。用户自定义参数（如这里的self.pipe）在__init__()方法中初始化；框架会把 Executor 构造期间传入的配置（uses_with、uses_metas等）透传给该构造方法。模型加载通常较重，放在__init__中意味着每个副本进程只会加载一次，推理时直接复用。

@requests 装饰器与端点机制

@requests def generate_image(self, docs: DocList[ImagePrompt], **kwargs) -> DocList[ImageDoc]: images = self.pipe(docs.text).images # image here is in PIL format for i, doc in enumerate(docs): doc.tensor = np.array(images[i])

任何被@requests装饰的 Executor 方法，都会在服务运行时通过对应的端点（endpoint）被调用。@requests的实现位于 jina/serve/executors/decorators.py#L219-L402，其核心规则是：

使用@requests(on='/foo')可把方法绑定到指定端点；
使用@requests(on=['/search', '/query'])可同时绑定多个端点；
裸@requests（不带on）会注册为默认回退处理器——任何未被其他方法绑定的端点请求都会落入该方法（见 jina/serve/executors/decorators.py#L241-L243）。

本例中generate_image使用裸@requests，因此无论客户端请求哪个端点，都会默认进入图像生成逻辑。此外，框架要求被装饰的方法必须带有**kwargs参数（见 jina/serve/executors/decorators.py#L301-L308），因为框架会在运行时注入路由信息、参数等额外关键字。装饰器还支持request_schema/response_schema显式声明输入输出类型，用于与 Pydantic 校验打通。

部署 Executor：使用 Deployment 提供服务

有了 Executor，接下来用Deployment把它变成真正可访问的服务。Deployment 可以在不修改任何 Executor 代码的前提下注入运行时配置，支持副本扩展、分片、动态批处理等能力。部署方式有两种：Python API 与 YAML 配置。

方式一：Python API

在deployment.py中：

from jina import Deployment dep = Deployment(uses=TextToImage, timeout_ready=-1) with dep: dep.block()

随后在命令行执行：

python deployment.py

timeout_ready=-1表示不限制 Executor 的就绪等待时间——这是加载大模型时的常用配置，因为StableDiffusionPipeline.from_pretrained首次加载可能耗时数分钟，若使用默认的就绪超时（源码默认timeout_ready=600000毫秒，见 jina/orchestrate/deployments/init.py#L304），可能因模型尚未加载完成而被判定启动失败。

方式二：YAML 配置

在deployment.yaml中：

jtype: Deployment with: uses: TextToImage py_modules: - text_to_image.py # name of the module containing your Executor timeout_ready: -1

然后通过 CLI 启动：

jina deployment --uses deployment.yml

py_modules列出 Executor 所在的 Python 模块，uses指定要加载的 Executor 类名；二者配合，Jina 会先导入text_to_image.py，再实例化其中的TextToImage类（该参数在源码中对应py_modules与uses两个构造参数，见 jina/orchestrate/deployments/init.py#L291-L311）。

启动成功后，终端会输出类似下面的端点信息：

──────────────────────────────────────── 🎉 Deployment is ready to serve! ───────────────────────────────────────── ╭────────────── 🔗 Endpoint ───────────────╮ │ ⛓ Protocol GRPC │ │ 🏠 Local 0.0.0.0:12345 │ │ 🔒 Private 172.28.0.12:12345 │ │ 🚪 Public 35.230.97.208:12345 │ ╰──────────────────────────────────────────╯

其中Protocol 默认为 GRPC（对应源码中protocol=['GRPC']的默认值，见 jina/orchestrate/deployments/init.py#L288），12345是本次服务绑定的端口。请注意Local / Private / Public三个地址分别表示容器/进程内、局域网与公网可达地址，客户端应使用与你运行环境匹配的那个端口。

在 Jupyter Notebook 等交互式环境中，不要在同一进程里先dep.block()再创建 Client 发请求——block()会阻塞当前线程。请把部署端与客户端拆分为不同进程/终端，或参考官方 Colab 的可复现代码。

编写 Client：向服务发送请求并接收结果

服务跑起来后，用jina.Client发送请求。客户端同样以 DocList 作为 IO 格式，且需要与 Executor 保持相同的输入 schema 定义：

from jina import Client from docarray import BaseDoc, DocList from docarray.documents import ImageDoc class ImagePrompt(BaseDoc): text: str image_prompt = ImagePrompt(text='rainbow unicorn butterfly kitten') client = Client(port=12345) # use port from output above response = client.post( on='/', inputs=DocListImagePrompt, return_type=DocList[ImageDoc], ) response[0].display()

Client(port=12345)指定与 Deployment 输出一致的端口；
client.post(on='/', ...)把请求发往根端点/——由于我们的 Executor 用的是裸@requests，任意端点（包括/）都会回退到generate_image方法（Client的post核心逻辑见 jina/clients/mixin.py#L344）；
inputs传入DocList[ImagePrompt]，return_type=DocList[ImageDoc]声明期望的返回类型，Jina 客户端据此完成反序列化。

在另一个终端执行：

python client.py

即可根据提示词rainbow unicorn butterfly kitten生成图像，response[0].display()会在 Notebook 中直接渲染结果：

图：客户端请求后生成的示例图像（提示词为 rainbow unicorn butterfly kitten）

扩展微服务：GPU、副本与动态批处理

单个 Executor 实例的吞吐有限。Jina 内置了开箱即用的扩展能力：replicas（副本）、shards（分片）与dynamic batching（动态批处理），可以在不改动 Executor 代码的前提下显著提升应用吞吐。

下面的 YAML 对上述 Deployment 做了三处升级（后续统一采用 YAML 方式，把部署逻辑与业务代码分离）：

jtype: Deployment with: timeout_ready: -1 uses: jinaai://jina-ai/TextToImage env: CUDA_VISIBLE_DEVICES: RR replicas: 2 uses_dynamic_batching: # configure dynamic batching /default: preferred_batch_size: 10 timeout: 200

各项配置的作用如下：

配置项	值	含义
`replicas`	`2`	启动两个 Executor 副本实例，负载均衡地处理请求
`env.CUDA_VISIBLE_DEVICES`	`RR`	为每个副本自动分配一张 GPU（`RR`表示 round-robin 依次分配，依次为 GPU 0、GPU 1……）
`uses_dynamic_batching`	见下	开启动态批处理，按端点配置批量聚合策略
`uses_dynamic_batching./default`	—	对默认端点生效的动态批处理规则
`preferred_batch_size`	`10`	目标批量大小：批处理器持续收集请求直到凑满 10 条，或达到超时阈值
`timeout`	`200`	最大等待时间（毫秒），队列中最老的请求等待达到 200ms 时，即使不足 10 条也会把当前批次发给 Executor

动态批处理的底层机制

动态批处理的核心思路是：把多个并发请求的 DocList 累积到队列中，按批一次性送入 Executor，从而摊薄模型推理开销、提升吞吐，代价是引入少量聚合延迟。该能力的底层实现在 jina/serve/executors/decorators.py#L405-L443 的@dynamic_batching装饰器中：

preferred_batch_size：目标批大小，批处理器会一直收集请求直到达到该数量，或达到timeout才触发；因此实际批大小 ≤preferred_batch_size；
timeout：单位为毫秒，默认 10_000ms（10 秒）。队列中最老的请求等待满timeout即强制成批发送；
更高级的参数还包括flush_all（超时触发时把队列中所有累积请求一次交付）、custom_metric/use_custom_metric（用自定义权重函数衡量批大小）等。

仓库的集成测试 tests/integration/dynamic_batching/test_dynamic_batching.py 对该机制做了系统性验证：既可以直接用@dynamic_batching(preferred_batch_size=4, timeout=2000)装饰 Executor 方法（见 tests/integration/dynamic_batching/test_dynamic_batching.py#L41-L49），也可以在 Deployment 层通过uses_dynamic_batching配置按端点注入（如{'/foo': {'preferred_batch_size': 2, 'timeout': 4000}}），两种方式的最终解析结果一致（见 tests/integration/dynamic_batching/test_dynamic_batching_config.py#L26-L40）。

图：多个副本副本间共享端点、均衡承接请求

说明与适用范围

假设宿主机有 2 张 GPU，使用上述扩展后的 YAML 会比单实例部署获得更高吞吐——replicas: 2每副本各占一张 GPU，动态批处理再把并发请求合并成批，双管齐下；
如果使用本教程开头的TextToImage类而非 Hub 上的jinaai://jina-ai/TextToImage，注意CUDA_VISIBLE_DEVICES: RR的 GPU 自动分配机制需要与 Jina 的副本启动流程配合，且要求本机 GPU 数量 ≥ 副本数；
GPU 相关的更多用法（gpus参数、--gpus all等）可参考 docs/tutorials/gpu-executor.md，Deployment 全量参数可查阅 docs/concepts/orchestration/deployment-args.md 与 YAML 规范 docs/yaml-spec.md。

得益于 YAML 语法，你可以在完全不触碰 Executor 代码的情况下注入副本数、GPU 环境变量、动态批处理等部署配置——当然，上述全部能力也都能通过 Python API 等价实现。从"模型"到"可扩展的 gRPC 微服务"，这就是 Jina 的标准路径。