In a model-serving API, an inference call may start work without immediately producing a prediction. Python’s async/await syntax lets the endpoint wait for that prediction before returning it, while the server remains able to handle other requests. Declare the endpoint with async def, call the runner’s async_run method, and await its result before using it. section 10.4.1
from bentoml.io import Image #1 from bentoml.io import JSON #1 #1 @svc.api(input=Image(), output=JSON()) #1 async def invocation(input_img): inf = yolo_v8_runner.inference.async_run([input_img]) #2 return await inf #3
The image-to-JSON endpoint starts YOLOv8 inference with async_run, then awaits the pending result before returning it. Without await, the endpoint could proceed with a result that is not yet available. section 10.4.1
Await each step that depends on earlier output. If an image must first be converted to grayscale, await that conversion before passing its result to detectors. When two detectors can work from the same converted image, asyncio.gather waits for both calls and supplies their results together. section 10.4.1
@svc.api(input=Image(), output=JSON())
async def predict(input_image: PIL.Image.Image) -> str:
model_input = await grayscale_converter.async_run(input_image)
results = await asyncio.gather(
object_detector_1.async_run(model_input),
object_detector_2.async_run(model_input),
)
return {"results": { "model_a": results[0], "model_b": results[1]
The endpoint awaits preprocessing, then uses asyncio.gather for two object-detection runners. The order expresses the dependency; the gathered calls express work that can proceed together. section 10.4.1
Async application code can also meet a synchronous testing interface. A small helper can run asynchronous initialization and query processing through asyncio.run(), then return only the response text needed by evaluation metrics. This keeps individual tests focused on response quality rather than async calling conventions. section 13.2.1
Listing 13.8 Helper to wrap async DakkaBot for sync testing
from dakka_bot_core import get_dakka_bot
import asyncio
dakka_bot = asyncio.run(get_dakka_bot()) #1
def get_response(query: str) -> str:
"""Helper function to get synchronous responses from async DakkaBot"""
try:
result = asyncio.run(dakka_bot.process_query(query)) #2
return result["response"] #3
except Exception:
return "" #4
The helper reuses one initialized bot across tests, calls its async query method from a synchronous function, and returns the response field. Its exception handler returns an empty string, so failures do not interrupt test execution. section 13.2.1
