Los agentes son entidades autónomas que perciben su entorno y actúan para alcanzar objetivos específicos. En mi artículo anterior, exploré los fundamentos de los agentes de IA y sus capacidades. En este artículo, profundizaremos en los sistemas multiagente.
Los sistemas multiagente se construyen cuando un agente individual no puede gestionar una tarea compleja. Podemos definirlos como una red de agentes que trabajan juntos para resolver un problema. Existen varias arquitecturas de agentes, y para nuestra implementación, nos centraremos en la arquitectura supervisora. Para impulsar nuestros agentes, utilizaremos Qwen, un modelo de lenguaje local extenso (LLM).
En este tutorial, crearemos un sistema de asistente de salud con IA compuesto por tres agentes: un agente de fitness, un agente de dietista y un agente de salud mental. Estos tres agentes trabajarán juntos para mejorar la condición física, la nutrición y el bienestar del usuario. El sistema estará coordinado por un agente supervisor, que asigna tareas a cada uno de estos agentes y supervisa su progreso.
Tabla de contenidos
La arquitectura del supervisor
En el centro del sistema se encuentra el agente supervisor, que desempeña el papel de coordinador. Recibe la información del usuario, identifica al agente o agentes adecuados para cada parte de la tarea y se la asigna. Por ejemplo, si un usuario dice: «Quiero ponerme en forma y comer más sano», el supervisor delega esta solicitud tanto al agente de fitness como al de dietista. A continuación, recopila sus respuestas y proporciona retroalimentación coherente al usuario. En el diagrama a continuación, podemos ver cómo cada agente se dirige al supervisor.
Ahora que hemos analizado el concepto y la arquitectura de nuestro Sistema Supervisor Multiagente, profundicemos en la implementación. Sigue el proceso paso a paso a continuación.
Paso 1: Instalación
Antes de poder usar Qwen localmente, necesitamos instalar Ollama, que nos permite ejecutar grandes modelos de lenguaje directamente en nuestro equipo.
i) Descargar e instalar Ollama
Para descargar Ollama, visita su sitio web oficial y descarga la versión compatible con tu sistema operativo.
ii) Verificar la instalación
ollama -v
ii) Extrae el modelo Qwen
ollama pull qwen2.5:14b
Paso 2: Configurar las claves API
Nuestro sistema se basará en API externas para proporcionar datos reales al Agente de Fitness y al Agente de Dietista. Estos agentes obtendrán información sobre ejercicios y nutrición de sus respectivas API.
API utilizadas:
- Agente de fitness: API de ejercicios de Ninjas
- Agente de dietética: API de alimentos y nutrición de Spoonacular
i) Obtén tus claves API
Regístrate para obtener cuentas gratuitas en ambas plataformas y recupera tus claves API.
ii) Claves de almacenamiento
Para mantener nuestras credenciales seguras y fácilmente accesibles, las almacenaremos en un archivo .env.
El archivo .env se verá así;
EXERCISE_API_KEY =xxxxxxxx
DIET_API_KEY =xxxxxxxxxxxx
Paso 3: Crear estado
Al crear nuestro Asistente de Salud con IA, uno de los primeros elementos que debemos configurar es el estado. Este juega un papel crucial para que nuestros agentes puedan realizar un seguimiento del historial de conversaciones mientras interactúan y se intercambian tareas durante el flujo de trabajo.
Para ello, usaremos la clase MessagesState integrada de LangGraph. Esta clase proporciona una forma práctica de almacenar y gestionar una lista de mensajes. Nuestra clase de estado personalizada heredará de MessageState para aprovechar su funcionalidad integrada.
from langchain_core.messages import HumanMessage,AIMessage
from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain.prompts import PromptTemplate
from IPython.display import display, Image
from typing import Annotated, Literal
from langchain_ollama import ChatOllama
from typing_extensions import TypedDict
from langchain.tools import tool
from langgraph.types import Command
import requests
import random
import uuid
import os
fitness_api_key = os.getenv("EXERCISE_API_KEY")
diet_api_key = os.getenv("DIET_API_KEY")
class State(MessagesState):
next: str
Paso 4: Crear herramientas personalizadas
Anteriormente, obtuvimos claves API de API-Ninjas (para datos de ejercicio) y Spoonacular (para datos de alimentación y nutrición). Ahora es el momento de usarlas creando herramientas personalizadas para nuestros agentes. Estas herramientas son las que el agente usará para realizar sus tareas.
i) Herramienta de fitness
Usaremos este endpoint para obtener varios tipos de ejercicio y generar una rutina de entrenamiento personalizada para los usuarios. Así es como se ve el código:
class FitnessData:
def __init__(self):
self.base_url = "https://api.api-ninjas.com/v1/exercises"
self.api_key = fitness_api_key
def get_muscle_groups_and_types(self):
muscle_targets = {
'full_body': ["abdominals", "biceps", "calves", "chest", "forearms", "glutes",
"hamstrings", "lower_back", "middle_back", "quadriceps",
"traps", "triceps", "adductors"
],
'upper_body': ["biceps", "chest", "forearms", "lats", "lower_back", "middle_back", "neck", "traps", "triceps" ],
'lower_body': ["adductors", "calves", "glutes", "hamstrings", "quadriceps"]
}
exercise_types = {'types':["powerlifting","strength", "stretching", "strongman"]}
return muscle_targets, exercise_types
def fetch_exercises(self, type, muscle, difficulty):
headers = {
'X-Api-Key':self.api_key
}
params= {
'type': type,
'muscle': muscle,
'difficulty': difficulty
}
try:
response = requests.get(self.base_url, headers=headers,params=params)
result = response.json()
if not result:
print(f"No exercises found for {muscle}")
return result
except requests.RequestException as e:
print(f"Request failed: {e}")
return []
def generate_workout_plan(self, query='full_body', difficulty='intermediate'):
output=[]
muscle_targets, exercise_types = self.get_muscle_groups_and_types()
muscle = random.choice(muscle_targets.get(query))
type = random.choice(exercise_types.get('types'))
result = self.fetch_exercises('stretching', muscle, difficulty)
print(result)
limit_plan = result[:3]
for i, data in enumerate(limit_plan):
if data not in output:
output.append(f"Exercise {i+1}: {data['name']}")
output.append(f"Muscle: {data['muscle']}")
output.append(f"Instructions: {data['instructions']}")
return output
Después, creamos la herramienta personalizada de fitness creando una instancia de la clase para llamar a la función generate_workout_plan. Esta función permite a los usuarios solicitar planes de entrenamiento basados en categorías específicas, como cuerpo completo, parte superior del cuerpo o parte inferior del cuerpo. Observarás el decorador @tool aplicado a la función; esto es lo que la convierte en una herramienta personalizada de LangChain.
@tool
def fitness_data_tool(query: Annotated[str, "This input will either be full_body, upper_body \
or lower_body exercise plan"]):
"""use this tool to get fitness or workout plan for a user.
The workout name provided serves as your input \
"""
fitness_tool = FitnessData()
result = fitness_tool.generate_workout_plan(query)
return result
ii) Herramienta para dietistas
Para la fuente de datos del agente dietista, utilizaremos la API de Spoonacular mediante sus puntos finales «Generar plan de comidas» y «Obtener información de recetas«. Con esto, el agente puede generar planes de comidas personalizados según las preferencias dietéticas de los usuarios, como vegetariano, vegano o dieta estándar. A partir de los resultados, los usuarios podrán ver los planes de comidas junto con un desglose nutricional diario, como proteínas, grasas y carbohidratos.
class Dietitian:
def __init__(self):
self.base_url = "https://api.spoonacular.com"
self.api_key = diet_api_key
def fetch_meal(self, time_frame="day", diet="None"):
url = f"{self.base_url}/mealplanner/generate"
params = {
"timeFrame":time_frame,
"diet": diet,
"apiKey":self.api_key
}
response = requests.get(url, params=params)
if not response:
print('Meal Plan not found')
return response.json()
def get_recipe_information(self, recipe_id):
url = f"{self.base_url}/recipes/{recipe_id}/information"
params = {"apiKey": self.api_key}
response = requests.get(url, params=params)
if not response:
print("Recipe not found")
return response.json()
def generate_meal_plan(self, query):
meals_processed = []
meal_plan = self.fetch_meal(query)
print(meal_plan)
meals = meal_plan.get('meals')
nutrients = meal_plan.get('nutrients')
for i, meal in enumerate(meals):
recipe_info = self.get_recipe_information(meal.get('id'))
ingredients = [ingredient['original'] for ingredient in recipe_info.get('extendedIngredients')]
meals_processed.append(f"🍽️ Meal {i+1}: {meal.get('title')}")
meals_processed.append(f"Prep Time: {meal.get('readyInMinutes')}")
meals_processed.append(f"Servings: {meal.get('servings')}")
meals_processed.append("📝 Ingredients:\n" + "\n".join(ingredients))
meals_processed.append(f"📋 Instructions:\n {recipe_info.get('instructions')}")
meals_processed.append(
"\n Daily Nutrients:\n"
f"Protein: {nutrients.get('protein', 'N/A')} kcal\n"
f"Fat: {nutrients.get('fat', 'N/A')} g\n"
f"Carbohydrates: {nutrients.get('carbohydrates', 'N/A')} g"
)
return meals_processed
A continuación, creamos nuestra herramienta personalizada a continuación;
@tool
def diet_tool(query: Annotated[str, "This input will either be None, vegetarian, and vegan"]):
"""use this tool to get diet plan for the user.
The diet type provided serves as your input \
"""
dietitian_tool = Dietitian()
result = dietitian_tool.generate_meal_plan(query)
return result
Paso 5: Definir LLM
Aquí, definiremos nuestro modelo de lenguaje extenso, que es el modelo Qwen2.5:14b. Este modelo es ideal para crear agentes inteligentes.
llm = ChatOllama(model="qwen2.5:14b")
memory = MemorySaver()
Paso 6: Creación del agente y los nodos
En este paso, crearemos nuestros nodos y agentes, utilizando la función create_react_agent preconstruida en LangGraph.
i) Agente de Fitness
Para el agente de fitness, pasamos tres componentes clave: el LLM, fitness_data_tool (nuestra herramienta personalizada para obtener datos de ejercicios) y fitness_agent_prompt a la función create_react_agent.
A continuación, en nuestro nodo de Fitness, que representa la tarea de fitness dentro del flujo de trabajo de LangGraph, invocamos al agente pasando el mensaje actual del estado (la entrada del usuario). Una vez que el agente procesa la entrada y genera una respuesta, pasamos el resultado mediante el objeto de comando. Esto nos ayuda a actualizar el estado con la salida e indicar al nodo de Fitness que regrese al agente supervisor una vez completada la tarea.
fitness_agent_prompt = """
You can only answer queries related to workout.
"""
fitness_agent = create_react_agent(
llm,
tools = [fitness_data_tool],
prompt = fitness_agent_prompt)
def fitness_node(state: State) -> Command[Literal["supervisor"]]:
result = fitness_agent.invoke(state)
return Command(
update={
"messages": [
AIMessage(content=result["messages"][-1].content, name="fitness")
]
},
goto="supervisor",
)
ii) Agente dietista
Para crear nuestro agente dietista y nuestro nodo, repetimos el mismo proceso.
dietitian_system_prompt = """
You can only answer queries related to diet and meal plans. .
"""
dietitian_agent = create_react_agent(
llm,
tools = [diet_tool],
prompt = dietitian_system_prompt)
def dietitian_node(state: State) -> Command[Literal["supervisor"]]:
result = dietitian_agent.invoke(state)
return Command(
update={
"messages": [
AIMessage(content=result["messages"][-1].content, name="dietitian")
]
},
goto="supervisor",
)
iii) Agente de Salud Mental
Para crear nuestro agente de salud mental, definimos un nodo de salud mental (mental_health_node) que incluye un mensaje personalizado para guiar al modelo de lenguaje completo sobre qué hacer y qué esperamos. Una vez completada la tarea, el nodo utiliza el objeto Comando para actualizar el estado de la conversación y luego redirige el control al Agente Supervisor.
def mental_health_node(state: State)-> Command[Literal["supervisor"]]:
prompt = PromptTemplate.from_template(
"""You are a supportive mental wellness coach.
Your task is to:
- Give a unique mental wellness tip or stress-reducing practice.
- Make it simple, kind, and useful. Avoid repeating tips."""
)
chain = prompt | llm
response = chain.invoke(state)
return Command(
update={
"messages": [
AIMessage(content=f"Here's your wellness tip: {response.content}", name="wellness")
]
},
goto="supervisor",
)
iv) Agente Supervisor
Al crear el Agente Supervisor, definimos un mensaje del sistema donde le indicamos su rol y presentamos el equipo que gestionará: el Agente de Fitness, el Agente de Dietista y el Agente de Salud Mental. También definimos una clase de Enrutador, que sirve como plantilla estructurada para la salida del supervisor.
Luego, implementamos el nodo supervisor, donde configuramos el flujo de mensajes y definimos la lógica de enrutamiento entre agentes. Esto incluye determinar cómo se enrutará a la siguiente tarea y cuándo concluir la conversación.
members = ["fitness", "dietitian", "wellness"]
options = members + ["FINISH"]
system_prompt = (
"You are a supervisor tasked with managing a conversation between the"
f" following workers: {members}. Given the following user request,"
" respond with the worker to act next. Each worker will perform a"
" task and respond with their results and status. When finished,"
" respond with FINISH."
"Guidelines:\n"
"1. Always check the last message in the conversation to determine if the task has been completed.\n"
"2. If you already have the final answer or outcome, return 'FINISH'.\n"
)
class Router(TypedDict):
"""Worker to route to next. If no workers needed, route to FINISH."""
next: Literal[*options]
def supervisor_node(state: State)-> Command[Literal[*members, "__end__"]]:
messages = [
{"role": "system", "content": system_prompt},
] + state["messages"]
response = llm.with_structured_output(Router).invoke(messages)
goto = response["next"]
if goto == "FINISH":
goto = END
return Command(goto=goto, update={"next": goto})
Paso 7: Crear un grafo multiagente
Ahora, creamos el grafo de flujo de trabajo, donde añadimos el nodo supervisor como punto de inicio de la ejecución. Después, añadimos los nodos de agente restantes.
builder = StateGraph(State)
builder.add_edge(START, "supervisor")
builder.add_node("supervisor", supervisor_node)
builder.add_node("fitness", fitness_node)
builder.add_node("dietitian", dietitian_node)
builder.add_node("wellness", mental_health_node)
graph = builder.compile(checkpointer=memory)
Paso 8: Probar el sistema multiagente
En esta etapa, nuestro sistema multiagente está completamente configurado y listo para recibir la entrada del usuario. Antes de enviar la entrada, definamos una función auxiliar para extraer la salida de los agentes.
def parse_langgraph_output(stream):
results = []
for key, value in stream.items():
if key == "supervisor":
continue
messages = value.get("messages", [])
for msg in messages:
if isinstance(msg, str):
results.append((key, msg))
elif isinstance(msg, AIMessage):
results.append((key, msg.content))
return results
Luego, pasamos la entrada del usuario al sistema.
# Get the final step in the stream
final_event = None
config = {"configurable": {"thread_id": "1", "recursion_limit": 10}}
inputs = {
"messages": [
HumanMessage(
content="Give me wellness tips for the month?"
)
],
}
for step in graph.stream(inputs, config=config):
final_event = step # Keep updating to the latest step
print(final_event)
response_message=parse_langgraph_output(final_event)
for agent, content in response_message:
print(f"**Agent :** `{agent}`\n\n{content}")
print("="*50)
Así es como se ve el resultado en la aplicación Streamlit.
Consulta este repositorio de GitHub para ver el código completo.
¡Gracias por leer! Nos vemos en la próxima.