Jina-Serve

Jina-serve 是一个用于构建和部署通过 gRPC、HTTP 和 WebSocket 进行通信的 AI 服务的框架。您可以在专注于核心逻辑的同时，将服务从本地开发扩展到生产环境。

核心特性

• 原生支持所有主流的机器学习框架和数据类型
• 高性能的服务设计，具备可扩展性、流式处理和动态批处理功能
• 支持流式输出的大语言模型推理
• 内置 Docker 集成和 Executor Hub
• 一键部署至 Jina AI Cloud
• 企业级支持，兼容 Kubernetes 和 Docker Compose

与 FastAPI 的对比

相较于 FastAPI 的主要优势：

• 基于 DocArray 的数据处理，并原生支持 gRPC
• 内置容器化和服务编排功能
• 无缝扩展微服务
• 一命令即可完成云端部署

安装

pip install jina

请参阅适用于 Apple Silicon 和 Windows 的安装指南。

核心概念

三个主要层次：

• 数据：BaseDoc 和 DocList 用于输入输出
• 服务层：Executor 处理 Documents，Gateway 连接各个服务
• 编排层：Deployments 提供 Executor 服务，Flows 创建工作流

构建 AI 服务

让我们使用 StableLM 创建一个基于 gRPC 的 AI 服务：

from jina import Executor, requests
from docarray import DocList, BaseDoc
from transformers import pipeline


class Prompt(BaseDoc):
    text: str


class Generation(BaseDoc):
    prompt: str
    text: str


class StableLM(Executor):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.generator = pipeline(
            'text-generation', model='stabilityai/stablelm-base-alpha-3b'
        )

    @requests
    def generate(self, docs: DocList[Prompt], **kwargs) -> DocList[Generation]:
        generations = DocList[Generation]()
        prompts = docs.text
        llm_outputs = self.generator(prompts)
        for prompt, output in zip(prompts, llm_outputs):
            generations.append(Generation(prompt=prompt, text=output))
        return generations

可通过 Python 或 YAML 部署：

from jina import Deployment
from executor import StableLM

dep = Deployment(uses=StableLM, timeout_ready=-1, port=12345)

with dep:
    dep.block()

jtype: Deployment
with:
 uses: StableLM
 py_modules:
   - executor.py
 timeout_ready: -1
 port: 12345

使用客户端：

from jina import Client
from docarray import DocList
from executor import Prompt, Generation

prompt = Prompt(text='suggest an interesting image generation prompt')
client = Client(port=12345)
response = client.post('/', inputs=[prompt], return_type=DocList[Generation])

构建工作流

将多个服务串联成一个 Flow：

from jina import Flow

flow = Flow(port=12345).add(uses=StableLM).add(uses=TextToImage)

with flow:
    flow.block()

扩展与部署

本地扩展

借助内置功能提升吞吐量：

• 使用副本实现并行处理
• 使用分片进行数据分区
• 动态批处理以提高模型推理效率

以下是一个对 Stable Diffusion 部署进行扩展的示例：

jtype: Deployment
with:
 uses: TextToImage
 timeout_ready: -1
 py_modules:
   - text_to_image.py
 env:
  CUDA_VISIBLE_DEVICES: RR
 replicas: 2
 uses_dynamic_batching:
   /default:
     preferred_batch_size: 10
     timeout: 200

云端部署

容器化服务

1. 整理您的 Executor：

TextToImage/
├── executor.py
├── config.yml
├── requirements.txt

2. 配置：

# config.yml
jtype: TextToImage
py_modules:
 - executor.py
metas:
 name: TextToImage
 description: Text to Image generation Executor

3. 推送至 Hub：

jina hub push TextToImage

部署至 Kubernetes

jina export kubernetes flow.yml ./my-k8s
kubectl apply -R -f my-k8s

使用 Docker Compose

jina export docker-compose flow.yml docker-compose.yml
docker-compose up

JCloud 部署

只需一条命令即可部署：

jina cloud deploy jcloud-flow.yml

LLM 流式处理

启用逐 token 流式输出，以构建响应迅速的 LLM 应用程序：

1. 定义 Schema：

from docarray import BaseDoc


class PromptDocument(BaseDoc):
    prompt: str
    max_tokens: int


class ModelOutputDocument(BaseDoc):
    token_id: int
    generated_text: str

2. 初始化服务：

from transformers import GPT2Tokenizer, GPT2LMHeadModel


class TokenStreamingExecutor(Executor):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.model = GPT2LMHeadModel.from_pretrained('gpt2')

3. 实现流式处理：

@requests(on='/stream')
async def task(self, doc: PromptDocument, **kwargs) -> ModelOutputDocument:
    input = tokenizer(doc.prompt, return_tensors='pt')
    input_len = input['input_ids'].shape[1]
    for _ in range(doc.max_tokens):
        output = self.model.generate(**input, max_new_tokens=1)
        if output[0][-1] == tokenizer.eos_token_id:
            break
        yield ModelOutputDocument(
            token_id=output[0][-1],
            generated_text=tokenizer.decode(
                output[0][input_len:], skip_special_tokens=True
            ),
        )
        input = {
            'input_ids': output,
            'attention_mask': torch.ones(1, len(output[0])),
        }

4. 服务端与客户端：

# 服务端
with Deployment(uses=TokenStreamingExecutor, port=12345, protocol='grpc') as dep:
    dep.block()


# 客户端
async def main():
    client = Client(port=12345, protocol='grpc', asyncio=True)
    async for doc in client.stream_doc(
        on='/stream',
        inputs=PromptDocument(prompt='what is the capital of France ?', max_tokens=10),
        return_type=ModelOutputDocument,
    ):
        print(doc.generated_text)

支持

Jina-serve 由 Jina AI 提供支持，并采用 Apache-2.0 许可证授权。

Jina-Serve 快速上手指南

Jina-Serve 是一个用于构建和部署 AI 服务的框架，支持 gRPC、HTTP 和 WebSocket 通信。它帮助开发者专注于核心逻辑，轻松将服务从本地开发扩展至生产环境。

环境准备

• 系统要求：支持 Linux、macOS（包括 Apple Silicon M1/M2）和 Windows。
  • Apple Silicon 用户请参考：安装指南
  • Windows 用户请参考：安装指南
• 前置依赖：
  • Python 3.7+
  • pip 包管理工具
• 网络建议：由于涉及模型下载（如 Hugging Face），国内用户建议配置相应的网络加速或镜像源以确保安装顺畅。

安装步骤

使用 pip 直接安装核心库：

pip install jina

提示：如果下载速度较慢，可使用国内镜像源加速安装：

pip install jina -i https://pypi.tuna.tsinghua.edu.cn/simple

基本使用

以下示例展示如何快速构建一个基于 StableLM 的 gRPC AI 服务，包含定义数据结构、创建执行器（Executor）、部署服务以及客户端调用。

1. 定义服务逻辑 (executor.py)

创建一个 Python 文件，定义输入输出数据结构及处理逻辑：

from jina import Executor, requests
from docarray import DocList, BaseDoc
from transformers import pipeline


class Prompt(BaseDoc):
    text: str


class Generation(BaseDoc):
    prompt: str
    text: str


class StableLM(Executor):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.generator = pipeline(
            'text-generation', model='stabilityai/stablelm-base-alpha-3b'
        )

    @requests
    def generate(self, docs: DocList[Prompt], **kwargs) -> DocList[Generation]:
        generations = DocList[Generation]()
        prompts = docs.text
        llm_outputs = self.generator(prompts)
        for prompt, output in zip(prompts, llm_outputs):
            generations.append(Generation(prompt=prompt, text=output))
        return generations

2. 部署服务

你可以选择直接使用 Python 代码或 YAML 配置文件进行部署。

方式 A：使用 Python 部署

from jina import Deployment
from executor import StableLM

dep = Deployment(uses=StableLM, timeout_ready=-1, port=12345)

with dep:
    dep.block()

方式 B：使用 YAML 部署 (config.yml)

jtype: Deployment
with:
 uses: StableLM
 py_modules:
   - executor.py
 timeout_ready: -1
 port: 12345

运行命令：jina deploy config.yml

3. 客户端调用

服务启动后，使用客户端发送请求并获取结果：

from jina import Client
from docarray import DocList
from executor import Prompt, Generation

prompt = Prompt(text='suggest an interesting image generation prompt')
client = Client(port=12345)
response = client.post('/', inputs=[prompt], return_type=DocList[Generation])

print(response[0].text)

---

更多高级功能（如流水线编排、Kubernetes 部署、LLM 流式输出等）请参阅官方完整文档。