Una guía de codificación para crear un mercado de agentes múltiples completamente funcional utilizando uAgent

En este tutorial, exploramos cómo construir un sistema multiagente pequeño pero funcional utilizando el marco uAgents. Configuramos tres agentes (Directorio, Vendedor y Comprador) que se comunican a través de protocolos de mensajes bien definidos para simular una interacción en el mercado del mundo real. Diseñamos esquemas de mensajes, definimos comportamientos de agentes e implementamos ciclos de solicitud-respuesta para demostrar el descubrimiento, la negociación y las transacciones entre agentes, todo ello ejecutándose de forma asincrónica en un bucle de eventos compartido. A través de esto, entendemos cómo los agentes autónomos colaboran, comercian y mantienen eficientemente flujos de trabajo descentralizados. Consulte los códigos completos aquí.

!pip -q install “uagents>=0.11.2” import asyncio, aleatorio al escribir import List, Dict, Opcional de uagents import Agente, Contexto, Oficina, Modelo, Clase de protocolo ServiceAnnounce(Modelo): categoría: str punto final: str clase ServiceQuery(Modelo): categoría: str clase ServiceList(Modelo): direcciones: Lista[str]

class OfertaRequest(Modelo): artículo: str max_price: int class Oferta(Modelo): artículo: str precio: int cant: int class Orden(Modelo): artículo: str cant: int class Recibo(Modelo): artículo: str cant: int total: int ok: bool nota: Opcional[str] = Ninguno

Comenzamos instalando la biblioteca uAgents y definiendo todos los modelos de mensajes que sustentan nuestro sistema de comunicación. Creamos tipos de datos estructurados para anuncios, consultas, ofertas y pedidos, lo que permite a los agentes intercambiar información sin problemas. Consulte los códigos completos aquí.

registro_proto = Protocolo(nombre=”registro”, versión=”1.0″) trade_proto = Protocolo(nombre=”comercio”, versión=”1.0″) directorio = Agente(nombre=”directorio”, semilla=”dir-seed-001″) vendedor = Agente(nombre=”vendedor”, semilla=”vendedor-seed-001″) comprador = Agente(nombre=”comprador”, seed=”buyer-seed-001″) directorio.include(registry_proto) seller.include(trade_proto) comprador.include(registry_proto) comprador.include(trade_proto) @registry_proto.on_message(model=ServiceAnnounce) async def on_announce(ctx: Context, sender: str, msg: ServiceAnnounce): reg = await ctx.storage.get(“reg”) o {} reg.setdefault(msg.category, set()).add(sender) await ctx.storage.set(“reg”, reg) ctx.logger.info(f”{remitente} registrado en ‘{msg.category}'”) @registry_proto.on_message(model=ServiceQuery) async def on_query(ctx: Context, sender: str, msg: ServiceQuery): reg = await ctx.storage.get(“reg”) o {} addrs = sorted(list(reg.get(msg.category, set()))) await ctx.send(sender, ServiceList(addresses=addrs)) ctx.logger.info(f”Proveedores {len(addrs)} devueltos para ‘{msg.category}'”)

Configuramos los agentes de Directorio, Vendedor y Comprador y definimos el protocolo de registro que gestiona el descubrimiento de servicios. Hacemos que el directorio responda a anuncios y consultas, permitiendo a los agentes registrarse y ubicarse entre sí de forma dinámica. Consulte los códigos completos aquí.

CATÁLOGO: Dictado[str, Dict[str, int]]= { “camera”: {“price”: 120, “qty”: 3}, “laptop”: {“price”: 650, “qty”: 2}, “auriculares”: {“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(“El vendedor anunció al directorio”) @trade_proto.on_message(model=OfferRequest) async def on_offer_request(ctx: Context, sender: str, req: OfferRequest): item = CATALOG.get(req.item) si no es un artículo: await ctx.send(sender, Offer(item=req.item, price=0, qty=0)) precio de devolución = max(1, int(item)[“price”] * (0.9 + 0.2 * random.random()))) si precio > req.max_price o artículo[“qty”] <= 0: espera ctx.send(remitente, oferta(artículo=req.item, precio=0, cantidad=0)) retorno espera ctx.send(remitente, oferta(item=req.item, precio=precio, cantidad=artículo)["qty"])) ctx.logger.info(f"Se ofreció {req.item} a {price} con cantidad {item['qty']}") @trade_proto.on_message(model=Order) async def on_order(ctx: Context, sender: str, order: Order): item = CATALOG.get(order.item) si no es artículo o artículo["qty"]

Creamos el catálogo del agente vendedor e implementamos lógica para responder a solicitudes de oferta y procesar pedidos. Simulamos el comercio del mundo real agregando precios variables y gestión de existencias, mostrando cómo el vendedor negocia y completa las transacciones. Consulte los códigos completos aquí.

@buyer.on_event(“startup”) async def comprador_start(ctx: Context): ctx.logger.info(“El comprador consulta el directorio de productos electrónicos…”) resp = await ctx.ask(directory.address, ServiceQuery(category=”electronics”), expects=ServiceList, timeout=5.0) sellers = resp.addresses si resp else []
si no son vendedores: objetivo de devolución = vendedores[0]
deseado = “laptop” presupuesto = 700 ctx.logger.info(f”Solicitando oferta para ‘{desired}’ dentro del presupuesto {budget} de {target}”) oferta = await ctx.ask(target, OfferRequest(item=desired, max_price=budget), expects=Offer, timeout=5.0) si no es oferta u oferta.precio <= 0: cantidad devuelta = 1 si oferta.qty >= 1 else 0 si cantidad == 0: return ctx.logger.info(f”Realizar pedido por {cantidad} x {oferta.artículo} a {oferta.precio}”) recibo = esperar ctx.ask(objetivo, pedido(artículo=oferta.artículo, cantidad=cantidad), espera=Recibo, tiempo de espera=5.0) si recibo y recibo.ok: ctx.logger.info(f”ÉXITO DEL PEDIDO: {recibo.cantidad} x {recibo.item} | total={recibo.total}”)

Programamos al agente Comprador para descubrir vendedores, solicitar ofertas y realizar pedidos según disponibilidad y presupuesto. Observamos cómo el comprador interactúa con el vendedor mediante comunicación asincrónica para completar una compra con éxito. Consulte los códigos completos aquí.

@buyer.on_interval(period=6.0) async def period_discovery(ctx: Context): visto = esperar ctx.storage.get(“visto”) o 0 si se ve >= 1: regresar en espera ctx.storage.set(“visto”, visto + 1) ctx.logger.info(“tick de descubrimiento periódico -> volver a consultar directorio”) resp = esperar ctx.ask(directorio.dirección, ServiceQuery(categoría=”electrónica”), expects=ServiceList, timeout=3.0) n = len(resp.direcciones) si resp else 0 ctx.logger.info(f”Periódico: informes de directorio {n} vendedor(es)”) oficina = Oficina() oficina.add(directorio) oficina.add(vendedor) Bureau.add(buyer) async def run_demo(segundos=10): tarea = asyncio.create_task(bureau.run_async()) intente: espere asyncio.sleep(segundos) finalmente: task.cancel() intente: espere tarea excepto asyncio.CancelledError: pase print(“\n✅ Ejecución de demostración completa.\n”) intente: bucle = asyncio.get_running_loop() espera run_demo(10) excepto RuntimeError: asyncio.run(run_demo(10))

Agregamos descubrimiento periódico para que el comprador vuelva a verificar a los vendedores disponibles y luego hacemos que la Oficina gestione a todos los agentes juntos. Lanzamos el tiempo de ejecución asincrónico para ver cómo se desarrolla y completa sin problemas la simulación completa del mercado.

En conclusión, hemos visto a nuestros agentes descubrirse unos a otros, negociar una oferta y completar una transacción completamente a través de interacciones basadas en mensajes. Nos damos cuenta de cómo uAgents simplifica la orquestación de múltiples agentes al combinar la estructura, la comunicación y la gestión del estado a la perfección dentro de Python. A medida que ejecutamos este ejemplo, no solo somos testigos de un sistema dinámico y autónomo en acción, sino que también obtenemos información sobre cómo la misma arquitectura puede extenderse a mercados descentralizados complejos, colaboraciones de IA y redes de servicios inteligentes, todo dentro de un marco liviano y fácil de usar.

Consulte los códigos completos aquí. No dude en consultar nuestra página de GitHub para tutoriales, códigos y cuadernos. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.

Asif Razzaq es el director ejecutivo de Marktechpost Media Inc.. Como emprendedor e ingeniero visionario, Asif está comprometido a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, Marktechpost, que se destaca por su cobertura en profundidad del aprendizaje automático y las noticias sobre aprendizaje profundo que es técnicamente sólida y fácilmente comprensible para una amplia audiencia. La plataforma cuenta con más de 2 millones de visitas mensuales, lo que ilustra su popularidad entre el público.

🙌 Siga MARKTECHPOST: agréguenos como fuente preferida en Google.

Una guía de codificación para crear un mercado de agentes múltiples completamente funcional utilizando uAgent

ByEquipo de 7 minutos

By Equipo de 7 minutos

Related Post

Herramienta de envíos de agentes Hermes Búsqueda de MCP: las evaluaciones antrópicas muestran una ganancia de precisión del 49% al 74% en Opus 4

Claude Opus 4.8 ya está disponible en AWS

Liquid AI lanza LFM2.5-8B-A1B: un modelo MoE en el dispositivo con 8,3 mil millones de parámetros activos y 1,5 mil millones en total

You missed

Una nueva herramienta que rastrea cómo el cerebro elimina los desechos podría ofrecer pistas sobre el Alzheimer

ERC y Junts mantienen a Sánchez en el alambre

El creciente coste de encontrar el amor « Euro Weekly News

Stunna Girl rompe su silencio después de un reciente arresto y una foto policial viral