Sistema multiagente (Multi-Agent System)
Este capítulo presenta el sistema multiagente (Multi-Agent System).
El sistema multiagente completa tareas complejas mediante la colaboración de múltiples agentes.
Esta es la tecnología clave para construir sistemas de IA a gran escala, capaz de realizar funciones complejas que un solo agente no puede completar.
Fundamentos de la arquitectura multiagente
Los desafíos clave de un sistema multiagente incluyen tres aspectos.
Problema de división del trabajo: cómo asignar tareas de manera razonable a diferentes agentes.
Problema de comunicación: cómo transmiten información los agentes entre sí.
Problema de coordinación: cómo garantizar que las acciones de múltiples agentes sean consistentes y eficientes.
Patrones de arquitectura básica
Arquitectura jerárquica (Hierarchical)
La arquitectura jerárquica adopta una estructura de árbol, con un agente principal responsable de la orquestación.
El agente principal (Orchestrator) se encarga de descomponer tareas e integrar resultados.
El agente secundario (Subagent) se encarga de ejecutar tareas específicas.
Orchestrator (调度者)
|
+-- Agent A (专家A)
+-- Agent B (专家B)
+-- Agent C (专家C)
Arquitectura entre pares (Peer-to-Peer)
En la arquitectura entre pares, todos los agentes tienen el mismo estatus y se comunican y colaboran directamente.
Adecuada para escenarios donde los agentes necesitan interacciones frecuentes entre iguales.
Agent A <--> Agent B
| |
v v
Agent C <--> Agent D
División de roles y colaboración
Un sistema multiagente eficaz requiere definiciones claras de roles.
Cada agente debe tener su propia especialidad y ámbito de responsabilidad.
Los mecanismos de colaboración garantizan que múltiples agentes puedan trabajar de manera coordinada.
Definición de roles comunes
Planificador (Planner): descompone tareas complejas y elabora un plan de ejecución.
Ejecutor (Executor): llama a herramientas y realiza operaciones concretas.
Revisor (Reviewer): evalúa los resultados y sugiere mejoras.
Coordinador (Coordinator): gestiona la comunicación entre agentes y la asignación de tareas.
Implementación de código
Implementación básica de un sistema multiagente
"""
Clase básica de Agente
Cada agente tiene su propio rol, herramientas y lógica de ejecución.
"""
def __init__(self, role, llm, tools):
# Nombre del rol del agente
self.role = role
# Modelo de lenguaje asociado
self.llm = llm
# Lista de herramientas disponibles
self.tools = tools
def execute(self, task, context=None):
"""
Ejecutar tarea
:param task: Descripción de la tarea
:param context: Información de contexto
:return: Resultado de la ejecución
"""
# Construir el prompt
prompt = self.build_prompt(task, context)
# Generar respuesta
response = self.llm.generate(prompt)
# Si es necesario llamar a una herramienta
if response.needs_tool_call:
return self.execute_tool(response)
return response.content
def build_prompt(self, task, context):
"""Construye el prompt del agente"""
return f"Rol: {self.role}\nTarea: {task}\nContexto: {context}"
def execute_tool(self, response):
"""Realiza la llamada a la herramienta"""
tool_name = response.tool_name
tool_input = response.tool_input
# Encuentra la herramienta correspondiente en la lista de herramientas
tool = self.get_tool(tool_name)
# Ejecuta la herramienta
return tool.execute(tool_input)
class MultiAgentSystem:
"""
Sistema de colaboración multiagente
Responsable de gestionar la colaboración entre múltiples agentes.
"""
def __init__(self, agents, coordinator):
# Diccionario de agentes, la clave es el nombre del rol
self.agents = {a.role: a for a in agents}
# Agente coordinador
self.coordinator = coordinator
def solve(self, task):
"""
Resolver tareas complejas
Completar mediante la coordinación de múltiples agentes.
"""
# Primer paso: el coordinador analiza la tarea y decide qué agentes se necesitan
plan = self.coordinator.decompose(task)
# Segundo paso: distribuir la tarea a los agentes correspondientes
results = {}
for role, subtask in plan.subtasks.items():
# Obtener el agente con el rol correspondiente
agent = self.agents.get(role)
if agent:
# Ejecutar la subtarea
results[role] = agent.execute(subtask)
# Tercer paso: el coordinador integra los resultados
final_result = self.coordinator.integrate(results)
return final_result
class CoordinatorAgent:
"""Agente coordinador"""
def __init__(self, llm):
self.llm = llm
def decompose(self, task):
"""
Descomponer tareas
Analizar la tarea, decidir qué agentes se necesitan y cómo dividir el trabajo.
"""
prompt = f"""
Analiza la siguiente tarea, divídela en subtareas y asígnala a los agentes adecuados.
Tarea: {task}
Por favor, genera:
1. Qué agentes se necesitan (roles)
2. Qué subtarea gestiona cada agente
3. Las dependencias entre subtareas
Formato de salida: JSON
"""
response = self.llm.generate(prompt)
return Plan.parse(response)
def integrate(self, results):
"""
Integrar los resultados de múltiples agentes
"""
prompt = f"""
Integra los resultados de ejecución de los siguientes agentes y genera la respuesta final.
Lista de resultados:
{results}
Por favor, integra y genera el resultado final.
"""
return self.llm.generate(prompt)
Framework AutoGen
AutoGen es un framework de programación multiagente desarrollado por Microsoft.
Proporciona un conjunto de API simples para construir sistemas de conversación multiagente.
AutoGen simplifica en gran medida la complejidad de desarrollo de los sistemas multiagente.
Conceptos clave
AssistantAgent: agente asistente inteligente capaz de llamar a herramientas.
UserProxyAgent: representa el comportamiento del usuario y puede ejecutar código y llamadas a herramientas.
GroupChat: admite la colaboración en chat grupal entre múltiples agentes.
GroupChatManager: gestiona las interacciones de los agentes en el chat grupal.
Ejemplos de código
Uso básico de AutoGen
import autogen
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
# ==================== Configurar LLM ====================
# Crear un diccionario de configuración
llm_config = {
"model": "gpt-4",
"api_key": "your-api-key", # En uso real, leer desde variables de entorno
"temperature": 0.7
}
# ==================== Crear un solo Agente ====================
# Crear agente asistente
assistant = AssistantAgent(
name="assistant",
system_message="""
Eres un útil asistente de programación en Python.
Puedes ayudar a los usuarios a escribir, depurar y optimizar código.
""",
llm_config=llm_config
)
# Crear agente proxy de usuario
# human_input_mode="NEVER" significa que no se requiere entrada humana
# max_consecutive_auto_reply=10 significa un máximo de 10 respuestas automáticas consecutivas
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER",
max_consecutive_auto_reply=10
)
# ==================== Conversación de un solo Agente ====================
# Iniciar conversación
user_proxy.initiate_chat(
assistant,
message="Ayúdame a escribir un algoritmo de ordenamiento rápido"
)
# ==================== Chat grupal multiagente ====================
# Crear múltiples agentes
coder = AssistantAgent(
name="coder",
system_message="Eres un experto en programación Python, encargado de escribir código.",
llm_config=llm_config
)
reviewer = AssistantAgent(
name="reviewer",
system_message="Eres un experto en revisión de código, encargado de revisar la calidad del código.",
llm_config=llm_config
)
# Crear chat grupal
group_chat = GroupChat(
agents=[coder, reviewer],
messages=[],
max_round=10 # Máximo 10 rondas de conversación
)
# Crear gestor de chat grupal
manager = GroupChatManager(
groupchat=group_chat,
llm_config=llm_config
)
# Iniciar chat grupal
user_proxy.initiate_chat(
manager,
message="Escribe un algoritmo de ordenamiento rápido y revisa el código"
)
Ejemplo más complejo con AutoGen
Llamada a herramientas con AutoGen
from autogen import AssistantAgent, UserProxyAgent
# Definir lista de herramientas
# Estas herramientas se registrarán en la lista de herramientas del agente
tools = [
{
"name": "calculate",
"description": "Realizar cálculo matemático",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Expresión matemática a calcular, como '2+3*5'"
}
},
"required": ["expression"]
}
},
{
"name": "search_web",
"description": "Buscar información web",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Palabra clave de búsqueda"
}
},
"required": ["query"]
}
}
]
# Crear un agente con capacidad de llamada a herramientas
assistant = AssistantAgent(
name="assistant",
system_message="""
Eres un asistente útil.
Cuando necesites calcular o buscar información, puedes invocar las herramientas correspondientes.
""",
llm_config={
"model": "gpt-4",
"api_key": "your-api-key"
},
tools=tools # Registrar herramienta
)
# Crear agente de usuario
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER"
)
# Iniciar la conversación, el Agente invocará automáticamente las herramientas
user_proxy.initiate_chat(
assistant,
message="¿Cuánto es (15 + 25) * 2?"
)
# Ejemplo de respuesta:
# El Agente podría responder:
# "Déjame calcularlo: (15 + 25) * 2 = 40 * 2 = 80"
# O invocar directamente la herramienta calculate para obtener el resultado
Protocolos A2A y MCP
Los sistemas multi-agente necesitan protocolos de comunicación estandarizados.
A2A y MCP son dos estándares de protocolo importantes.
Protocolo A2A (Agent-to-Agent)
El protocolo A2A define el formato estándar de comunicación entre agentes.
Admite descubrimiento de agentes, colaboración de tareas y sincronización de estado.
Funciones principales
Descubrimiento de agentes (Agent Discovery): Los agentes pueden descubrir las capacidades y servicios de otros agentes.
Colaboración de tareas (Task Collaboration): Múltiples agentes pueden colaborar para completar tareas complejas.
Sincronización de estado (State Synchronization): Los agentes pueden sincronizar información de estado entre sí.
Protocolo MCP (Model Context Protocol)
MCP es un protocolo estándar abierto.
Permite que los modelos de IA se conecten de forma segura con herramientas externas y fuentes de datos.
MCP es el protocolo estándar para la integración de herramientas, similar a la función de una interfaz USB.
Componentes principales
Host: Entorno host que ejecuta aplicaciones de IA.
Client: Cliente que establece conexión con el servidor MCP.
Server: Servidor que proporciona herramientas y recursos.
Ejemplo de definición de herramienta MCP
Definición de herramientas MCP
"name": "filesystem",
"description": "Herramienta de operaciones del sistema de archivos",
"version": "1.0.0",
"tools": [
{
"name": "read_file",
"description": "Leer contenido del archivo",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Ruta del archivo, obligatoria"
},
"encoding": {
"type": "string",
"description": "Codificación del archivo, opcional, por defecto utf-8"
}
},
"required": ["path"]
}
},
{
"name": "write_file",
"description": "Escribir contenido en el archivo",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Ruta del archivo, obligatoria"
},
"content": {
"type": "string",
"description": "Contenido del archivo, obligatorio"
}
},
"required": ["path", "content"]
}
},
{
"name": "list_directory",
"description": "Listar contenido del directorio",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Ruta del directorio"
}
}
}
}
],
"resources": [
{
"uri": "file://config",
"description": "Archivo de configuración"
}
]
}
Nota: A2A y MCP resuelven problemas diferentes. A2A resuelve el problema de comunicación entre agentes, mientras que MCP resuelve el problema de conexión entre agentes y herramientas externas. Ambos pueden usarse en conjunto para construir un sistema multi-agente completo.
Patrón maestro-esclavo de agentes
El patrón maestro-esclavo es un patrón clásico de arquitectura multi-agente.
El agente maestro (Orchestrator) es responsable de la descomposición de tareas y la integración de resultados.
El agente esclavo (Subagent) es responsable de ejecutar tareas específicas.
Escenarios de aplicación
Escenarios donde las tareas pueden descomponerse claramente en subtareas independientes.
Escenarios que requieren la colaboración de agentes con diferentes habilidades especializadas.
Flujos de trabajo complejos que requieren coordinación y control centralizados.
Implementación de código
Implementación del patrón maestro-esclavo
"""
Agente de orquestación principal (agente maestro)
Responsable de la descomposición de tareas, asignación e integración de resultados
"""
def __init__(self, llm, subagents):
# Modelo de lenguaje
self.llm = llm
# Diccionario de agentes esclavos
self.subagents = {a.role: a for a in subagents}
def handle_request(self, request):
"""
Procesar solicitudes del usuario
Flujo de orquestación completo
"""
# ==================== Fase de análisis ====================
# Analizar la solicitud, decidir si se necesita la colaboración de múltiples agentes
plan = self.llm.plan(request)
if len(plan.subtasks) == 1:
# Tarea única, distribuir directamente al agente correspondiente
target_role = plan.target_role
return self.subagents[target_role].execute(plan.subtasks[0])
# ==================== Fase de colaboración multi-agente ====================
# Crear cola de tareas
task_queue = self.create_task_queue(plan.subtasks)
# Almacenar resultados completados
results = []
# Bucle de procesamiento hasta que todas las tareas estén completadas
while task_queue:
# Obtener tareas que se pueden ejecutar en paralelo
parallel_tasks = self.get_parallel_tasks(task_queue)
# Ejecución en paralelo
parallel_results = self.execute_in_parallel(parallel_tasks)
results.extend(parallel_results)
# Actualizar cola de tareas
# Gestionar dependencias, eliminar tareas completadas, añadir nuevas tareas
task_queue = self.update_queue(task_queue, parallel_results)
# ==================== Fase de integración ====================
# Integrar todos los resultados
return self.integrate_results(results)
def create_task_queue(self, subtasks):
"""Crear cola de tareas"""
return TaskQueue(subtasks)
def get_parallel_tasks(self, task_queue):
"""
Obtener tareas que se pueden ejecutar en paralelo
Excluir tareas con dependencias
"""
ready_tasks = []
for task in task_queue.tasks:
# Verificar si todas las dependencias de las tareas están satisfechas
if all(dep in task_queue.completed for dep in task.dependencies):
ready_tasks.append(task)
return ready_tasks
def execute_in_parallel(self, tasks):
"""Ejecutar múltiples tareas en paralelo"""
import concurrent.futures
results = []
with concurrent.futures.ThreadPoolExecutor() as executor:
# Enviar todas las tareas
future_to_task = {
executor.submit(self.execute_task, task): task
for task in tasks
}
# Recopilar resultados
for future in concurrent.futures.as_completed(future_to_task):
task = future_to_task[future]
try:
result = future.result()
results.append(result)
except Exception as e:
results.append(TaskResult(task=task, error=str(e)))
return results
def execute_task(self, task):
"""Ejecutar una sola tarea"""
agent = self.subagents.get(task.role)
if not agent:
return TaskResult(task=task, error=f"Unknown role: {task.role}")
return agent.execute(task.description)
def update_queue(self, task_queue, completed_results):
"""Actualizar cola de tareas"""
# Marcar tareas completadas como finalizadas
for result in completed_results:
task_queue.completed.add(result.task.id)
# Eliminar tareas completadas
task_queue.tasks = [
t for t in task_queue.tasks
if t.id not in task_queue.completed
]
return task_queue
def integrate_results(self, results):
"""
Integrar los resultados de múltiples agentes
Generar respuesta final
"""
prompt = f"""
Integra los resultados de ejecución de los siguientes agentes y genera una respuesta final.
Lista de resultados:
{results}
Por favor, integra en una respuesta completa y coherente.
"""
return self.llm.generate(prompt)
class Subagent:
"""Clase base del agente esclavo"""
def __init__(self, role, llm, tools=None):
self.role = role
self.llm = llm
self.tools = tools or []
def execute(self, task):
"""Lógica concreta de ejecución de tareas"""
raise NotImplementedError
Ventajas y desventajas del patrón maestro-esclavo
| Ventajas | Desventajas |
|---|---|
| Estructura clara, fácil de entender e implementar | El agente maestro se convierte en un punto único de fallo |
| Coordinación centralizada, fácil de controlar | El agente maestro puede convertirse en un cuello de botella de rendimiento |
| Adecuado para tareas con jerarquía clara | Menos flexible, difícil de ajustar dinámicamente |
| Depuración conveniente, se puede rastrear cada subtarea | No adecuado para escenarios con interacciones peer-to-peer frecuentes |
Resumen del capítulo
Este capítulo presenta los conceptos centrales y los métodos de implementación de los sistemas multi-agente.
Fundamentos de la arquitectura multi-agentePresenta dos modos básicos: arquitectura jerárquica y arquitectura de igual a igual.
División de roles y colaboraciónMejora la eficiencia del sistema mediante definiciones claras de roles.
Marco de trabajo AutoGenProporciona un marco de programación que simplifica el desarrollo multi-agente.
Protocolos A2A y MCPSon protocolos estándar para la comunicación de agentes y la integración de herramientas.
Patrón de agente maestro-esclavoAdecuado para tareas complejas con una jerarquía clara.
Elegir la arquitectura multi-agente adecuada depende del escenario específico.
Para escenarios simples, un solo agente es suficiente.
Para escenarios complejos, es necesario elegir el patrón de arquitectura adecuado según las características de la tarea.
Otras extensiones