Создание модульных событийно-управляемых AI-агентов с UAgents и Google Gemini на Python

Установка окружения и зависимостей

В этом уроке показано, как построить легковесную архитектуру событийно-управляемых AI-агентов с использованием фреймворка UAgents поверх API Google Gemini. Начинаем с установки необходимых библиотек:

!pip install -q uagents google-genai

Флаг q обеспечивает тихую установку, чтобы вывод оставался чистым. Этот набор инструментов позволяет запускать событийно-управляемых AI-агентов с Gemini.

Подготовка окружения Python

Импортируем необходимые модули и применяем nest_asyncio, чтобы включить вложенные циклы событий — это важно для запуска асинхронных процессов в интерактивных средах, например, в ноутбуках.

import os, time, multiprocessing, asyncio
import nest_asyncio  
from google import genai
from pydantic import BaseModel, Field
from uagents import Agent, Context
 
nest_asyncio.apply()

Импорты включают системные утилиты, клиент Google GenAI, Pydantic для валидации данных и основные классы UAgents.

Конфигурация клиента Gemini API

Устанавливаем ключ API Gemini в переменные окружения и инициализируем клиент GenAI для аутентификации запросов на генерацию контента.

os.environ["GOOGLE_API_KEY"] = "Введите Ваш API ключ здесь"
 
client = genai.Client()

Замените плейсхолдер на ваш реальный ключ, чтобы получить доступ.

Определение схем сообщений с помощью Pydantic

Определяем структурированные форматы сообщений для обмена между агентами с использованием моделей Pydantic.

class Question(BaseModel):
    question: str = Field(...)
 
class Answer(BaseModel):
    answer: str = Field(...)

Эти модели гарантируют, что сообщения валидны и хорошо структурированы.

Создание агента Gemini

Агент gemini_agent слушает входящие вопросы, вызывает модель Gemini "flash" для получения ответов и отправляет ответы обратно.

ai_agent = Agent(
    name="gemini_agent",
    seed="agent_seed_phrase",
    port=8000,
    endpoint=["http://127.0.0.1:8000/submit"]
)
 
@ai_agent.on_event("startup")
async def ai_startup(ctx: Context):
    ctx.logger.info(f"{ai_agent.name} слушает на {ai_agent.address}")
 
def ask_gemini(q: str) -> str:
    resp = client.models.generate_content(
        model="gemini-2.0-flash",
        contents=f"Answer the question: {q}"
    )
    return resp.text
 
@ai_agent.on_message(model=Question, replies=Answer)
async def handle_question(ctx: Context, sender: str, msg: Question):
    ans = ask_gemini(msg.question)
    await ctx.send(sender, Answer(answer=ans))

Агент регистрирует обработчики запуска и сообщений, интегрируясь с Gemini для динамической генерации ответов.

Создание клиентского агента

Агент client_agent при запуске отправляет вопрос и обрабатывает полученный ответ, выводя его и корректно завершая цикл событий.

client_agent = Agent(
    name="client_agent",
    seed="client_seed_phrase",
    port=8001,
    endpoint=["http://127.0.0.1:8001/submit"]
)
 
@client_agent.on_event("startup")
async def ask_on_start(ctx: Context):
    await ctx.send(ai_agent.address, Question(question="What is the capital of France?"))
 
@client_agent.on_message(model=Answer)
async def handle_answer(ctx: Context, sender: str, msg: Answer):
    print(" Ответ от Gemini:", msg.answer)
    # Корректное завершение
    asyncio.create_task(shutdown_loop())
 
async def shutdown_loop():
    await asyncio.sleep(1)  # Время на очистку
    loop = asyncio.get_event_loop()
    loop.stop()

Этот агент демонстрирует инициацию общения и управление жизненным циклом.

Совместный запуск агентов

Используем multiprocessing Python для одновременного запуска агентов, обеспечивая правильный порядок запуска и завершения.

def run_agent(agent):
    agent.run()
 
if __name__ == "__main__":
    p = multiprocessing.Process(target=run_agent, args=(ai_agent,))
    p.start()
    time.sleep(2)  
 
    client_agent.run()
 
    p.join()

gemini_agent запускается в отдельном процессе, а client_agent — в основном потоке, что позволяет асинхронно обмениваться сообщениями.

Итоги

Данный урок предоставляет модульный шаблон для создания событийно-управляемых AI-агентов с помощью UAgents и Google Gemini. Рассмотрены настройка окружения, создание агентов, обработка сообщений и параллельное выполнение, что позволяет создавать сложные AI-решения с минимальными усилиями.