В этом руководстве мы рассмотрим, как создать небольшую, но функциональную мультиагентную систему с использованием фреймворка uAgents. Мы настроим трёх агентов — Directory, Seller и Buyer — которые будут взаимодействовать через чётко определённые протоколы обмена сообщениями, чтобы имитировать взаимодействие на реальном рынке.
Шаг 1: установка библиотеки uAgents
Сначала мы устанавливаем библиотеку uAgents и определяем все модели сообщений, которые лежат в основе нашей системы связи. Мы создаём структурированные типы данных для объявлений, запросов, предложений и заказов, что позволяет агентам беспрепятственно обмениваться информацией.
«`python
!pip -q install «uagents>=0.11.2»
import asyncio, random
from typing import List, Dict, Optional
from uagents import Agent, Context, Bureau, Model, Protocol
«`
Шаг 2: определение моделей сообщений
«`python
class ServiceAnnounce(Model):
category: str
endpoint: str
class ServiceQuery(Model):
category: str
class ServiceList(Model):
addresses: List[str]
class OfferRequest(Model):
item: str
max_price: int
class Offer(Model):
item: str
price: int
qty: int
class Order(Model):
item: str
qty: int
class Receipt(Model):
item: str
qty: int
total: int
ok: bool
note: Optional[str] = None
«`
Шаг 3: настройка агентов
«`python
registry_proto = Protocol(name=»registry», version=»1.0″)
trade_proto = Protocol(name=»trade», version=»1.0″)
directory = Agent(name=»directory», seed=»dir-seed-001″)
seller = Agent(name=»seller», seed=»seller-seed-001″)
buyer = Agent(name=»buyer», seed=»buyer-seed-001″)
directory.include(registry_proto)
seller.include(trade_proto)
buyer.include(registry_proto)
buyer.include(trade_proto)
«`
Шаг 4: обработка сообщений
«`python
@registryproto.onmessage(model=ServiceAnnounce)
async def on_announce(ctx: Context, sender: str, msg: ServiceAnnounce):
reg = await ctx.storage.get(«reg») or {}
reg.setdefault(msg.category, set()).add(sender)
await ctx.storage.set(«reg», reg)
ctx.logger.info(f»Registered {sender} under ‘{msg.category}'»)
@registryproto.onmessage(model=ServiceQuery)
async def on_query(ctx: Context, sender: str, msg: ServiceQuery):
reg = await ctx.storage.get(«reg») or {}
addrs = sorted(list(reg.get(msg.category, set())))
await ctx.send(sender, ServiceList(addresses=addrs))
ctx.logger.info(f»Returned {len(addrs)} providers for ‘{msg.category}'»)
«`
Мы создали агентов Directory, Seller и Buyer и определили протокол реестра, который управляет обнаружением служб. Мы заставили Directory отвечать на объявления и запросы, позволяя агентам динамически регистрироваться и находить друг друга.
Шаг 5: реализация логики продавца
«`python
CATALOG: Dict[str, Dict[str, int]] = {
«camera»: {«price»: 120, «qty»: 3},
«laptop»: {«price»: 650, «qty»: 2},
«headphones»: {«price»: 60, «qty»: 5},
}
@seller.on_event(«startup»)
async def seller_start(ctx: Context):
await ctx.send(directory.address, ServiceAnnounce(category=»electronics», endpoint=seller.address))
ctx.logger.info(«Seller announced to directory»)
@tradeproto.onmessage(model=OfferRequest)
async def onofferrequest(ctx: Context, sender: str, req: OfferRequest):
item = CATALOG.get(req.item)
if not item:
await ctx.send(sender, Offer(item=req.item, price=0, qty=0))
return
price = max(1, int(item[«price»] (0.9 + 0.2 random.random())))
if price > req.max_price or item[«qty»] <= 0:
await ctx.send(sender, Offer(item=req.item, price=0, qty=0))
return
await ctx.send(sender, Offer(item=req.item, price=price, qty=item[«qty»]))
ctx.logger.info(f»Offered {req.item} at {price} with qty {item[‘qty’]}»)
«`
Мы создали каталог продавца и реализовали логику для ответа на запросы предложений и обработки заказов. Мы смоделировали реальную торговлю, добавив переменное ценообразование и управление запасами, показывая, как продавец ведёт переговоры и завершает транзакции.
Шаг 6: реализация логики покупателя
«`python
@buyer.on_event(«startup»)
async def buyer_start(ctx: Context):
ctx.logger.info(«Buyer querying directory for electronics…»)
resp = await ctx.ask(directory.address, ServiceQuery(category=»electronics»), expects=ServiceList, timeout=5.0)
sellers = resp.addresses if resp else []
if not sellers:
return
target = sellers[0]
desired = «laptop»
budget = 700
ctx.logger.info(f»Requesting offer for ‘{desired}’ within budget {budget} from {target}»)
offer = await ctx.ask(target, OfferRequest(item=desired, max_price=budget), expects=Offer, timeout=5.0)
if not offer or offer.price <= 0:
return
qty = 1 if offer.qty >= 1 else 0
if qty == 0:
return
ctx.logger.info(f»Placing order for {qty} x {offer.item} at {offer.price}»)
receipt = await ctx.ask(target, Order(item=offer.item, qty=qty), expects=Receipt, timeout=5.0)
if receipt and receipt.ok:
ctx.logger.info(f»ORDER SUCCESS: {receipt.qty} x {receipt.item} | total={receipt.total}»)
«`
Мы запрограммировали агента-покупателя на поиск продавцов, запрос предложений и размещение заказов на основе доступности и бюджета. Мы наблюдаем, как покупатель взаимодействует с продавцом через асинхронную связь для успешного завершения покупки.
В заключение мы увидели, как наши агенты обнаруживают друг друга, договариваются о предложении и завершают транзакцию через взаимодействие на основе сообщений. Мы поняли, как uAgents упрощает координацию работы нескольких агентов, объединяя структуру, коммуникацию и управление состоянием в Python.
Проверьте полные коды здесь. Не стесняйтесь посетить нашу страницу GitHub для руководств, кодов и блокнотов. Также подписывайтесь на нас в Twitter и присоединяйтесь к нашему сообществу в ML SubReddit. Подписывайтесь на нашу рассылку. И присоединяйтесь к нам в Telegram!
1. Какие основные шаги включает в себя создание мультиагентной системы с использованием фреймворка uAgents?
Ответ: создание мультиагентной системы с использованием фреймворка uAgents включает в себя следующие основные шаги:
1. Установка библиотеки uAgents.
2. Определение моделей сообщений, которые лежат в основе системы связи.
3. Настройка агентов (Directory, Seller и Buyer).
4. Обработка сообщений между агентами.
5. Реализация логики продавца и покупателя.
2. Какие типы данных используются для обмена информацией между агентами в данной системе?
Ответ: для обмена информацией между агентами в данной системе используются следующие типы данных:
* ServiceAnnounce (объявления);
* ServiceQuery (запросы);
* ServiceList (списки адресов);
* OfferRequest (запросы предложений);
* Offer (предложения);
* Order (заказы);
* Receipt (чеки).
3. Какие протоколы используются для взаимодействия агентов в данной системе?
Ответ: для взаимодействия агентов в данной системе используются протоколы registry и trade. Протокол registry управляет обнаружением служб, а протокол trade используется для обмена предложениями и заказами между продавцом и покупателем.
4. Какие функции выполняет агент Directory в данной системе?
Ответ: агент Directory выполняет следующие функции:
* Регистрация агентов и их категорий.
* Обработка объявлений и запросов от других агентов.
* Предоставление списка адресов агентов, предоставляющих определённые услуги.
5. Какие инструменты и библиотеки используются для создания мультиагентной системы в данном руководстве?
Ответ: для создания мультиагентной системы в данном руководстве используется библиотека uAgents, а также следующие инструменты и библиотеки:
* Python;
* asyncio;
* random;
* typing;
* Context;
* Bureau;
* Model;
* Protocol.