В этом руководстве мы покажем, как использовать фреймворк UAgents для создания лёгкой архитектуры AI-агентов, управляемых событиями, на базе API Google Gemini.
Шаг 1: настройка вложенных циклов событий
Сначала мы применяем `nest_asyncio`, чтобы включить вложенные циклы событий, затем настраиваем ключ API Gemini и создаём экземпляр клиента GenAI.
Шаг 2: определение коммуникационных контрактов
Мы определяем наши коммуникационные контракты, модели вопроса и ответа Pydantic, а затем запускаем два UAgents: один «geminiagent», который прослушивает входящие сообщения с вопросами, вызывает модель Gemini «flash» для генерации ответов и выдаёт сообщения с ответами; и один «clientagent», который запускает запрос при запуске и обрабатывает входящий ответ.
Шаг 3: запуск агентов одновременно
Мы изучаем, как запустить этих агентов одновременно, используя утилиту Python multiprocessing, и корректно завершаем работу цикла событий после завершения обмена, иллюстрируя плавную координацию UAgents для обмена сообщениями между агентами.
«`python
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()
os.environ[«GOOGLEAPIKEY»] = «Use Your Own API Key Here»
client = genai.Client()
class Question(BaseModel):
question: str = Field(…)
class Answer(BaseModel):
answer: str = Field(…)
ai_agent = Agent(
name=»gemini_agent»,
seed=»agentseedphrase»,
port=8000,
endpoint=[«http://127.0.0.1:8000/submit»]
)
@aiagent.onevent(«startup»)
async def ai_startup(ctx: Context):
ctx.logger.info(f»{aiagent.name} listening on {aiagent.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
@aiagent.onmessage(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))
client_agent = Agent(
name=»client_agent»,
seed=»clientseedphrase»,
port=8001,
endpoint=[«http://127.0.0.1:8001/submit»]
)
@clientagent.onevent(«startup»)
async def askonstart(ctx: Context):
await ctx.send(ai_agent.address, Question(question=»What is the capital of France?»))
@clientagent.onmessage(model=Answer)
async def handle_answer(ctx: Context, sender: str, msg: Answer):
print(» Answer from Gemini:», msg.answer)
# Use a more graceful shutdown
asyncio.createtask(shutdownloop())
async def shutdown_loop():
await asyncio.sleep(1) # Give time for cleanup
loop = asyncio.geteventloop()
loop.stop()
def run_agent(agent):
agent.run()
if name == «main«:
p = multiprocessing.Process(target=runagent, args=(aiagent,))
p.start()
time.sleep(2)
client_agent.run()
p.join()
«`
С помощью этого руководства, сфокусированного на UAgents, мы получили чёткий план создания модульных AI-сервисов, которые взаимодействуют через чётко определённые хуки событий и схемы сообщений. Вы увидели, как UAgents упрощает управление жизненным циклом агентов, регистрацию событий запуска, обработку входящих сообщений и отправку структурированных ответов, без необходимости написания шаблонного сетевого кода.
Теперь вы можете расширить свою настройку UAgents, включив в неё более сложные рабочие процессы разговора, несколько типов сообщений и динамическое обнаружение агентов.