Desvendando a GenAI - Parte 13
Explorando Multi-Agentes Distribuídos: Um Exemplo Simples para Resolver Problemas Complexos
A ideia é ter dois agentes: um que recebe um texto e gera um resumo, e outro que lê os resumos e toma uma decisão. Toda a comunicação entre os agentes é feita por filas (neste caso, para simplificar, estamos usando uma biblioteca baseada em SQLite).
Antes de entrar no código é importante saber um pouquinho sobre interação entre agentes…
Tipos de interação entre Agentes
Comunicação em Rede (Network)
Na comunicação em rede, todos os agentes se comunicam diretamente uns com os outros sem uma hierarquia definida. Esse tipo de comunicação é descentralizado e permite que os agentes compartilhem informações e colaborem de maneira mais flexível. É útil em sistemas onde a colaboração e a troca de informações entre todos os agentes são essenciais para o funcionamento do sistema.
Comunicação Supervisora (Supervisor)
Na comunicação supervisora, há um agente supervisor que coordena e controla as ações dos outros agentes. O supervisor recebe informações dos agentes subordinados, toma decisões com base nessas informações e envia instruções de volta aos agentes. Esse tipo de comunicação é útil em sistemas onde é necessário um controle centralizado para garantir que os agentes trabalhem de maneira coordenada e eficiente. Nesses sistemas é comum ter um controle de estado (stateful)
Comunicação Hierárquica (Hierarchical)
Na comunicação hierárquica, os agentes são organizados em uma estrutura de níveis, onde cada nível de agentes se comunica com o nível imediatamente superior ou inferior. Esse tipo de comunicação é estruturado e permite uma clara definição de responsabilidades e fluxos de informação. É útil em sistemas complexos onde diferentes níveis de agentes têm diferentes funções e responsabilidades.
Customizado
Cada agente se comunica com um ou mais agentes de forma livre.
Voltando pro nosso exemplo…
Nosso exemplo foi implementado utilizando interações customizadas entre agentes. De forma simplificada, o processo segue o seguinte fluxo:
Produtor gera um texto de entrada;
Fila de Resumo;
Agente especializado em Resumo pega o texto de entrada e gera um resumo;
Fila de Decisão;
Agente especializado em Decisão gera uma decisão baseada no resumo;
Fim.
Código
import os
import numpy as np
from openai import OpenAI
from dotenv import load_dotenv
import ell
import sys
import time
from litequeue import LiteQueue, Message
# Para executar usando OpenAI
# Carregando variáveis de ambiente a partir do arquivo .env
# load_dotenv(r".env")
# OPENAI_SERVICE_ACCOUNT_KEY = os.getenv("OPENAI_SERVICE_ACCOUNT_KEY")
# Configurar a chave da API
# client = OpenAI(api_key=OPENAI_SERVICE_ACCOUNT_KEY)
# Para executar usando Ollama (gratuito)
MODEL = "llama3.1"
client = OpenAI(
base_url="http://localhost:11434/v1",
api_key="ollama",
)
ell.config.verbose = True
ell.config.register_model(MODEL, client)
# LLM responsável pelo Resumo, será usado no agente "Resumidor"
@ell.simple(model="llama3.1", client=client)
def analyse_and_summary(text: str):
return [
ell.system("You are an especialist agent that analyse text and sumarizes."),
ell.user(f"Analyze the following data and provide a summary: {text}"),
]
# LLM responsável pela Decisão, será usado no agente "Decisor"
@ell.simple(model="llama3.1", client=client)
def take_action(summary: str):
return [
ell.system("You are an especialist agent that makes decision based on summaries."),
ell.user(f"Based on the following summary: '{summary}', what is the best decision to make?"),
]
if __name__ == "__main__":
args = sys.argv[1:]
print(args)
db_path = "queue.sqlite3"
# O ideal aqui era que cada um desses trechos fossem separados em arquivos. Porém, para facilitar o tutorial, coloquei tudo em um único arquivo.
# Trecho que representa uma aplicação produtora ("Producer")
if len(args) == 0:
print("--- Init start Producer ---")
data = r"Data shows that sales increased by 20% in the last quarter due to a new marketing campaign."
q1 = LiteQueue(db_path, queue_name="topic1")
q1.put(data)
# Trecho que representa nosso agente responsável pela sumarização
elif args == "summarizer":
print("--- Init Summarizer ---")
q1 = LiteQueue(db_path, queue_name="topic1")
q2 = LiteQueue(db_path, queue_name="topic2")
while True:
if q1.qsize() == 0:
continue
data: Message = q1.pop()
if data is None:
continue
summary = analyse_and_summary(str(data.data))
q2.put(summary)
time.sleep(0.1)
# Trecho que representa nosso agente responsável pela decisão
elif args == "actioner":
print("--- Init Actioner ---")
q2 = LiteQueue(db_path, queue_name="topic2")
while True:
if q2.qsize() == 0:
continue
data: Message = q2.pop()
if data is None:
continue
action = take_action(str(data.data))
print(action)
time.sleep(0.1)Executando o Exemplo
Para rodar o exemplo, abra 3 terminais (consoles). No primeiro, execute:
python 10-MultiAgent.py actioner
No segundo, execute:
python 10-MultiAgent.py summarizer
No terceiro, execute:
python 10-MultiAgent.py
O código fonte pode ser encontrado aqui!
Conclusão
Este exemplo simples serve como uma base para a criação de sistemas mais complexos e distribuídos, onde a colaboração entre agentes pode ser adaptada para atender a diversas necessidades e cenários.
Se tiver alguma dúvida ou precisar de mais informações, sinta-se à vontade para perguntar.
Deixo a implementação das outras interações com vocês…
E ai curtiu? Me ajude se inscrevendo!