调用 API 端点

BentoML 提供了客户端实现，允许您向 BentoML 服务 (Services) 发送同步和异步请求。

本文档解释了如何使用 BentoML 客户端 (clients)。

客户端类型

根据您的需求，可以使用以下类创建 BentoML 客户端对象。

• bentoml.SyncHTTPClient: 定义同步客户端，适用于简单、阻塞式操作，即应用程序等待响应后再继续执行。

• bentoml.AsyncHTTPClient: 定义异步客户端，适用于非阻塞式操作，允许应用程序在等待响应时处理其他任务。

创建客户端

假设您的 BentoML 服务有一个名为 summarize 的端点，该端点接受字符串 text 作为输入，并返回文本的摘要版本，如下所示。

class Summarization:
    def __init__(self) -> None:
        # Load model into pipeline
        self.pipeline = pipeline('summarization')

    @bentoml.api
    def summarize(self, text: str) -> str:
        result = self.pipeline(text)
        return result[0]['summary_text']

启动 Summarization 服务后，可以通过指定服务器地址来创建以下客户端。

同步

import bentoml

client = bentoml.SyncHTTPClient('https://:3000')
summarized_text: str = client.summarize(text="Your long text to summarize")
print(summarized_text)

# Close the client to release resources
client.close()

异步

import asyncio
import bentoml

async def async_client_operation():
    async with bentoml.AsyncHTTPClient('https://:3000') as client:
        summarized_text: str = await client.summarize(text="Your long text to summarize")
        print(summarized_text)

asyncio.run(async_client_operation())

在上述同步和异步客户端中，请求被发送到托管在 https://:3000 的服务的 summarize 端点。BentoML 客户端实现支持与服务 API 对应的方法，这些方法应使用与服务中定义的相同参数（在本例中为 text）调用。这些方法是根据服务的端点动态创建的，提供了到服务功能的直接映射。

在此示例中，客户端上的 summarize 方法直接映射到 Summarization 服务中的 summarize 方法。传递给 summarize 方法的数据（text="Your long text to summarize"）符合服务的预期输入。

如果您为 API 端点定义了根输入 (root input)，则在客户端中仅使用位置参数，而无需指定参数名称。

注意

如果您将服务部署到 BentoCloud，可以使用 get_client() 或 get_async_client() 获取部署的客户端。更多信息请参阅与部署交互 (Interact with the Deployment)。

使用上下文管理器

为了增强资源管理并减少连接泄漏的风险，建议您在上下文管理器中创建客户端，如下所示。

同步

import bentoml

with bentoml.SyncHTTPClient('https://:3000') as client:
    summarized_text: str = client.summarize(text="Your long text to summarize")
    print(summarized_text)

异步

import bentoml

async with bentoml.AsyncHTTPClient('https://:3000') as client:
    summarized_text: str = await client.summarize(text="Your long text to summarize")
    print(summarized_text)

检查服务就绪状态

在调用特定的服务方法之前，可以使用客户端的 is_ready 方法检查服务是否已准备好处理请求。这确保了您的 API 调用仅在服务启动并运行时进行。

import bentoml

client = bentoml.SyncHTTPClient('https://:3000')
if client.is_ready():
    summarized_text: str = client.summarize(text="Your long text to summarize.")
    print("Summarized text:", summarized_text)
else:
    print("Service is not ready")

client.close()

另外，可以使用 server_ready_timeout 参数指定客户端在超时前等待 BentoML 服务就绪的最长秒数。这在初次连接可能正在启动的服务时非常有用。如果服务在指定超时时间内未就绪，客户端将引发超时异常。

import bentoml

client = bentoml.SyncHTTPClient(
  'https://:3000',
  server_ready_timeout=60  # Wait up to 60 seconds for the Service to be ready
)
summarized_text: str = client.summarize(text="Your long text to summarize")
print(summarized_text)

client.close()

调用任务端点

您可以创建客户端来与定义了任务 (task) 端点的服务进行交互，通过提交输入并在稍后异步检查结果。这对于客户端无需主动等待任务完成的场景特别有用。更多信息请参阅异步任务队列 (Async task queues)。

输入和输出

BentoML 客户端支持处理不同的输入和输出类型。

JSON

使用 BentoML 的 HTTP 客户端可以轻松处理 JSONable 数据输入和 JSON 输出，这些客户端旨在无缝序列化和反序列化 JSON 数据。

对于输入，当您发送可以序列化为 JSON 的数据（例如，字典、列表、字符串和数字）时，只需将其作为参数传递给与您的服务 API 对应的客户端方法。

以下代码来自此示例项目的 SentenceEmbedding 服务，该服务接受 JSONable 输入（在本例中为列表）。

import typing as t

@bentoml.service
class SentenceEmbedding:
    ...

    @bentoml.api
    def encode(self, sentences: t.List[str] = SAMPLE_SENTENCES) -> np.ndarray:
    ...

要创建用于处理像 SentenceEmbedding 这样的服务的 JSONable 输入的客户端：

import bentoml
import typing as t

client = bentoml.SyncHTTPClient("https://:3000")

# Specify the sentences for the request
sentences_list: t.List[str] = [
    "The sun dips below the horizon, painting the sky orange.",
    "A gentle breeze whispers through the autumn leaves.",
    "The moon casts a silver glow on the tranquil lake.",
    # Add more if necessary
]

# Make the request using the Service endpoint
result = client.encode(sentences=sentences_list)

# Print the result
print(f"Encoded sentences result: {result}")

client.close()

对于输出，当 BentoML 服务返回 JSON 数据时，客户端会自动将此 JSON 反序列化为 Python 数据结构（例如字典或列表，具体取决于 JSON 结构）。

以下代码来自此示例项目的 WhisperX 服务，该服务返回 JSONable 输出（在本例中为字典）。

import typing as t
from pathlib import Path

@bentoml.service
class WhisperX:
    ...

    @bentoml.api
    def transcribe(self, audio_file: Path) -> t.Dict:
    ...

要创建用于处理像 WhisperX 这样的服务的 JSONable 输出的客户端：

import bentoml
import typing as t

client = bentoml.SyncHTTPClient('https://:3000')

# Set the audio URL
audio_url = 'https://example.org/female.wav'

# The response is expected to be a dictionary
response: t.Dict = client.transcribe(audio_file=audio_url)

print(response)

提示

您可以打印 JSON 响应中特定键的值。例如，WhisperX 服务返回以下内容，您可以输出第一个 segment 的文本：

response = {
    "segments": [
        {
            "start": 0.009,
            "end": 2.813,
            "text": " The Hispaniola was rolling scuppers under in the ocean swell.",
            "words": [
                {"word": "The", "start": 0.009, "end": 0.069, "score": 0.0},
                {"word": "Hispaniola", "start": 0.109, "end": 0.81, "score": 0.917},
                # Other words omitted...
            ],
        },
        # Other segments omitted...
    ],
    "word_segments": [
        {"word": "The", "start": 0.009, "end": 0.069, "score": 0.0},
        {"word": "Hispaniola", "start": 0.109, "end": 0.81, "score": 0.917},
        # Other words omitted...
    ],
}

# Print the text of the first segment
# Add the following line to your client code
print("Segment text:", response["segments"][0]["text"])

文件

BentoML 客户端支持各种文件类型，例如图像和通用二进制文件。

对于文件输入，您传递指向文件的 Path 对象。客户端负责读取文件并将其作为请求的一部分发送。对于文件输出，客户端将输出作为 Path 对象提供。您可以使用此 Path 对象来访问、读取或处理文件。

以下代码片段来自ControlNet 示例，该示例接受并返回图像文件。

import PIL
from PIL.Image import Image as PIL_Image

@bentoml.service
class ControlNet:
    ...

    @bentoml.api
    async def generate(self, image: PIL_Image, params: Params) -> PIL_Image:
    ...

要创建用于处理像 ControlNet 这样的服务的文件输入和输出的客户端：

import bentoml
from pathlib import Path

client = bentoml.SyncHTTPClient("https://:3000")

# Specify the image path and other parameters for the request
image_path: Path = Path("/path/to/example-image.png")
params = {
    "prompt": "A young man walking in a park, wearing jeans.",
    "negative_prompt": "ugly, disfigured, ill-structure, low resolution",
    "controlnet_conditioning_scale": 0.5,
    "num_inference_steps": 25
}

# Make the request using the Service endpoint
result_path: Path = client.generate(
    image=image_path,
    params=params,
)

print(f"Generated file saved at: {result_path}")

client.close()

您也可以使用 URL 作为输入，如下所示：

import bentoml
from pathlib import Path

client = bentoml.SyncHTTPClient("https://:3000")

# Specify the image URL and other parameters for the request
image_url = 'https://example.org/1.png'
# The remaining code is the same
...

流式处理

您可以向 BentoML 客户端添加流式处理逻辑，这在处理大量数据或实时数据流时特别有用。根据客户端类型，流式输出会返回一个 generator 或 async generator。

同步

对于同步流式处理，SyncHTTPClient 使用 Python generator 从流中接收数据并输出。

import bentoml

client = bentoml.SyncHTTPClient("https://:3000")
for data_chunk in client.stream_data():
    # Process each chunk of data as it arrives
    process_data(data_chunk)

client.close()

def process_data(data_chunk):
    # Add processing logic
    print("Processing data chunk:", data_chunk)
    # Add more logic here to handle the data chunk

异步

对于异步流式处理，AsyncHTTPClient 使用 async generator。这允许对流式数据进行异步迭代。

import bentoml

async with bentoml.AsyncHTTPClient("https://:3000") as client:
    async for data_chunk in client.stream_data():
        # Process each chunk of data as it arrives
        await process_data_async(data_chunk)

async def process_data_async(data_chunk):
    # Add processing logic
    print("Processing data chunk asynchronously:", data_chunk)
    # Add more complex asynchronous processing here
    await some_async_operation(data_chunk)

授权

当使用需要身份验证的 BentoML 服务时，您可以使用令牌授权客户端（SyncHTTPClient 和 AsyncHTTPClient）。此令牌通常是 JWT (JSON Web Token) 或其他形式的 API 密钥，用于确保客户端被允许访问指定的 BentoML 服务。令牌包含在客户端发出的每个请求的 HTTP 头中，允许服务器验证客户端的凭据。

要授权客户端，您可以在初始化期间将令牌作为参数传递。

import bentoml

client = bentoml.SyncHTTPClient('https://:3000', token='your_token_here')
summarized_text: str = client.summarize(text="Your long text to summarize")
print(summarized_text)

client.close()

错误处理

处理错误、检查错误代码和消息以及实现重试对于可靠的客户端-服务器通信非常重要。以下是一些关于错误处理和重试的策略和示例。

基础知识

与 BentoML 服务交互时，可能会发生网络问题、服务停机或无效输入等错误。适当的错误处理允许您的客户端优雅地响应这些问题。

您可以使用 try 和 except 块来捕获请求期间可能发生的异常。

import bentoml
from bentoml.exceptions import BentoMLException

client = bentoml.SyncHTTPClient('https://:3000')

try:
    summarized_text: str = client.summarize(text="Your long text to summarize.")
    print(summarized_text)
except BentoMLException as e:
    print(f"An error occurred: {e}")
finally:
    client.close()

捕获异常时，检查特定的错误代码或消息以确定失败原因非常有用。这可以指导重试逻辑或更精确地通知您问题所在。

实现重试逻辑

重试失败的请求有助于克服诸如网络中断或服务不可用等临时问题。实现重试时，考虑使用指数退避（exponential backoff）以避免使服务器或网络过载。

以下是实现带有指数退避的重试的简单示例。

import time
from bentoml.exceptions import BentoMLException
import bentoml

def retry_request(client, max_retries=3, backoff_factor=2):
    for attempt in range(max_retries):
        try:
            summarized_text: str = client.summarize(text="Your long text to summarize.")
            return summarized_text
        except BentoMLException as e:
            print(f"Attempt {attempt+1}: An error occurred: {e}")
            time.sleep(backoff_factor ** attempt)
    print("Max retries reached. Giving up.")

client = bentoml.SyncHTTPClient('https://:3000')

try:
    response = retry_request(client)
    if response:
        print(response)
finally:
    client.close()