Copyright 2026 Google LLC.
# @title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.Gemini API: Асинхронные запросы Python
Изображение (внешний источник)
В этом блокноте будет показано, как выполнять асинхронные и параллельные запросы с использованием Python SDK Gemini API и стандартной библиотеки Python 3 asyncio.
Примеры здесь запускаются в Google Colab и используют неявный цикл событий, предоставляемый в Colab. Вы также можете запускать эти команды интерактивно, используя REPL asyncio (вызывается с помощью python -m asyncio), или вы можете управлять циклом событий самостоятельно.
Примечание: Этот блокнот использует Interactions API, новейший способ взаимодействия с моделями Gemini. Ищете версию
generateContent? Проверьте архивную ветку.
%pip install -U -q "google-genai>=2.9.0" aiohttp # 2.0 for Interactions APINote: you may need to restart the kernel to use updated packages.Настройка ключа API
Чтобы запустить следующую ячейку, ваш ключ API должен быть сохранен в секрете Colab с именем GEMINI_API_KEY. Если у вас еще нет ключа API или вы не знаете, как создать секрет Colab, см. Аутентификация для пошагового руководства.
from google.colab import userdata
from google import genai
GEMINI_API_KEY = userdata.get("GEMINI_API_KEY")
client = genai.Client(api_key=GEMINI_API_KEY)Выберите модель, которую вы хотите использовать в этом руководстве:
MODEL_ID = "gemini-3.8-flash" # @param ["gemini-4-argon", "gemini-3.1-pro-preview", "gemini-3.8-flash", "gemini-3.7-flash", "gemini-3.6-flash", "gemini-3.5-flash-lite", "gemini-2.5-pro"] {"allow-input": true, "isTemplate": true}Использование локальных файлов
Этот простой пример показывает, как вы можете использовать локальные файлы (предполагается, что они загружаются быстро) с асинхронным API SDK.
prompt = "Describe this image in just 3 words."
img_filenames = ["firefighter.jpg", "elephants.jpeg", "jetpack.jpg"]
img_dir = "https://storage.googleapis.com/generativeai-downloads/images/"Начните с загрузки файлов локально.
!wget -nv {img_dir}{{{','.join(img_filenames)}}}2026-06-22 11:49:41 URL:https://storage.googleapis.com/generativeai-downloads/images/firefighter.jpg [547369/547369] -> "firefighter.jpg" [1]2026-06-22 11:49:41 URL:https://storage.googleapis.com/generativeai-downloads/images/elephants.jpeg [224007/224007] -> "elephants.jpeg" [1]2026-06-22 11:49:41 URL:https://storage.googleapis.com/generativeai-downloads/images/jetpack.jpg [357568/357568] -> "jetpack.jpg" [1]
FINISHED --2026-06-22 11:49:41--
Total wall clock time: 0.6s
Downloaded: 3 files, 1.1M in 0.02s (59.4 MB/s)import PIL.Image
from IPython.display import display
for img_filename in img_filenames:
print(img_filename)
display(PIL.Image.open(img_filename).resize((300, 200)))firefighter.jpg<PIL.Image.Image image mode=RGB size=300x200>
elephants.jpeg<PIL.Image.Image image mode=RGB size=300x200>
jetpack.jpg<PIL.Image.Image image mode=RGB size=300x200>
Асинхронный код использует метод aio.interactions.create для вызова API. Большинство асинхронных методов API можно найти в пространстве имен aio.
Обратите внимание, что этот код не выполняется параллельно. Асинхронный вызов указывает, что цикл событий может передавать управление другим задачам, но в этом коде не запланировано никаких других задач. Этого может быть достаточно, например, если вы запускаете это в обработчике запросов веб-сервера, так как это позволит обработчику передавать управление другим задачам во время ожидания ответа API.
import PIL
import base64
from io import BytesIO
async def describe_local_images():
for img_filename in img_filenames:
# Convert image to base64 for the API
img = PIL.Image.open(img_filename)
buffered = BytesIO()
img.save(buffered, format="JPEG")
img_b64 = base64.b64encode(buffered.getvalue()).decode("utf-8")
interaction = await client.aio.interactions.create(
model=MODEL_ID,
input=[
{"type": "image", "data": img_b64, "mime_type": "image/jpeg"},
{"type": "text", "text": prompt},
],
)
print(interaction.steps[-1].content[0].text)
await describe_local_images()/usr/local/google/home/giom/nb-fixes/venv/lib/python3.13/site-packages/IPython/core/interactiveshell.py:3746: UserWarning: Async interactions client cannot use aiohttp, fallingback to httpx.
await eval(code_obj, self.user_global_ns, self.user_ns)Cat in tree.Wild elephant familyJetpack backpack sketchАсинхронная и параллельная загрузка изображений
Этот пример показывает более реальный случай, когда изображение загружается из внешнего источника с использованием асинхронной HTTP-библиотеки aiohttp, и каждое изображение обрабатывается параллельно.
import io, aiohttp, asyncio
async def download_image(session: aiohttp.ClientSession, img_url: str) -> bytes:
"""Returns base64-encoded image data from the provided URL."""
async with session.get(img_url) as img_resp:
data = await img_resp.read()
return base64.b64encode(data).decode("utf-8")
async def process_image(img_future: asyncio.Future) -> str:
"""Summarise the image using the Interactions API."""
img_b64 = await img_future
interaction = await client.aio.interactions.create(
model=MODEL_ID,
input=[
{"type": "image", "data": img_b64, "mime_type": "image/jpeg"},
{"type": "text", "text": prompt},
],
)
return interaction.steps[-1].content[0].textasync def download_and_describe():
async with aiohttp.ClientSession() as sesh:
response_futures = []
for img_filename in img_filenames:
# Create the image download tasks (this does not schedule them yet).
img_future = download_image(sesh, img_dir + img_filename)
# Kick off the Gemini API request using the pending image download tasks.
text_future = process_image(img_future)
# Save the reference so they can be processed as they complete.
response_futures.append(text_future)
print(f"Download and content generation queued for {len(response_futures)} images.")
# Process responses as they complete (may be a different order). The tasks are started here.
for response in asyncio.as_completed(response_futures):
print()
print(await response)
await download_and_describe()Download and content generation queued for 3 images.Jetpack backpack sketchWild elephant family
Firefighter rescues catВ приведенном выше примере для каждого изображения создается сопрограмма, которая загружает, а затем суммирует изображение. Сопрограммы выполняются на последнем шаге, в цикле as_completed. Чтобы запустить их как можно раньше, не блокируя другую работу, вы могли бы обернуть download_image в asyncio.ensure_future, но для этого примера выполнение было отложено, чтобы разделить вопросы создания и выполнения.
Следующие шаги
- Ознакомьтесь с классом
AsyncClientв справочнике Python SDK. - Подробнее о библиотеке Python
asyncio