ENJAMBRE_HERRAMIENTAS = [
{
“type”: “function”,
“function”: {
“name”: “task_update”,
“description”: “Update the status of a task. Use ‘in_progress’ when starting, ‘completed’ when done.”,
“parameters”: {
“type”: “object”,
“properties”: {
“task_id”: {“type”: “string”, “description”: “The task ID”},
“status”: {“type”: “string”, “enum”: [“in_progress”, “completed”, “failed”]}, “resultado”: {“tipo”: “cadena”, “descripción”: “Resultado o salida de la tarea”}, }, “obligatorio”: [“task_id”, “status”]}, }, }, { “type”: “function”, “function”: { “name”: “inbox_send”, “description”: “Enviar un mensaje a otro agente (p. ej., ‘líder’ o el nombre de un trabajador).”, “parameters”: { “type”: “object”, “properties”: { “to”: {“type”: “string”, “description”: “Nombre del agente destinatario”}, “message”: {“type”: “string”, “description”: “Contenido del mensaje”}, }, “obligatorio”: [“to”, “message”]}, }, }, { “type”: “function”, “function”: { “name”: “inbox_receive”, “description”: “Comprueba y consume todos los mensajes en tu bandeja de entrada.”, “parameters”: { “type”: “object”, “properties”: {}, }, }, }, { “type”: “function”, “function”: { “name”: “task_list”, “description”: “Enumera las tareas asignadas a ti o a todas las tareas del equipo.”, “parámetros”: { “tipo”: “objeto”, “properties”: { “propietario”: {“tipo”: “cadena”, “descripción”: “Filtrar por nombre de propietario (opcional)”}, }, }, }, }, ]clase SwarmAgent: def __init__( self, nombre: str, rol: str, system_prompt: str, task_board: TaskBoard, bandeja de entrada: InboxSystem, registro: TeamRegistry,): self.name = nombre self.role = rol self.system_prompt = system_prompt self.task_board = task_board self.inbox = bandeja de entrada self.registry = registro self.conversation_history: lista[dict] = []
self.inbox.register(nombre) self.registry.register(nombre, rol) def _build_system_prompt(self) -> str: coord_protocol = f””” ## Protocolo de coordinación (autoinyectado: eres el agente ‘{self.name}’) Eres parte de un enjambre de agentes de IA. Tu rol: {self.role} Tu nombre: {self.name} Herramientas disponibles (equivalentes a ClawTeam CLI): – task_list: verifique sus tareas asignadas (como `clawteam task list`) – task_update: actualice el estado de la tarea a in_progress/completed/failed (como `clawteam task update`) – inbox_send: envíe mensajes a otros agentes (como `clawteam inbox send`) – inbox_receive: revise su bandeja de entrada en busca de mensajes (como `clawteam inboxceived`) FLUJO DE TRABAJO: 1. Verifique sus tareas con task_list 2. Marque una tarea como en progreso cuando comienza 3. Haga el trabajo (piense, analice, produzca resultados) 4. Marque la tarea como completada con su resultado 5. Envíe un mensaje de resumen al ‘líder’ cuando haya terminado “”” return self.system_prompt + “\n” + coord_protocol def _handle_tool_call(self, nombre_herramienta: str, args: dict) -> str: if nombre_herramienta == “actualización_tarea”: estado = Estado de la tarea (argumentos)[“status”]) resultado = args.get(“resultado”, “”) self.task_board.update_status(args[“task_id”]estado, resultado) si estado == TaskStatus.COMPLETED: self.registry.increment_completed(self.name) return json.dumps({“ok”: True, “task_id”: args[“task_id”]”nuevo_estado”: argumentos[“status”]}) elif nombre_herramienta == “inbox_send”: self.inbox.send(self.nombre, argumentos[“to”]argumentos[“message”]) devuelve json.dumps({“ok”: Verdadero, “sent_to”: args[“to”]}) elif nombre_herramienta == “inbox_receive”: msgs = self.inbox.receive(self.name) si no es mensaje: return json.dumps({“messages”: []”nota”: “No hay mensajes nuevos”}) return json.dumps({ “mensajes”: [
{“from”: m.sender, “content”: m.content, “time”: m.timestamp}
for m in msgs
]
}) elif nombre_herramienta == “lista_tareas”: propietario = args.get(“propietario”, self.nombre) tareas = self.task_board.get_tasks(propietario=propietario) return json.dumps({“tareas”: [t.to_dict() for t in tasks]}) return json.dumps({“error”: f”Herramienta desconocida: {tool_name}”}) def run(self, user_message: str, max_iterations: int = 6) -> str: self.conversation_history.append({“role”: “user”, “content”: user_message}) para la iteración en el rango (max_iterations): intente: respuesta = client.chat.completions.create(modelo=MODELO, mensajes=[
{“role”: “system”, “content”: self._build_system_prompt()},
*self.conversation_history,
]herramientas=SWARM_TOOLS, tool_choice=”auto”, temperatura=0.4, ) excepto Excepción como e: devolver f”[API Error] {e}” elección = respuesta.opciones[0]
msg = elección.message asistente_msg = {“role”: “asistente”, “contenido”: msg.content o “”} si msg.tool_calls: asistente_msg[“tool_calls”] = [
{
“id”: tc.id,
“type”: “function”,
“function”: {“name”: tc.function.name, “arguments”: tc.function.arguments},
}
for tc in msg.tool_calls
]
self.conversation_history.append(assistant_msg) si no es msg.tool_calls: devuelve msg.content o “(Sin respuesta)” para tc en msg.tool_calls: fn_name = tc.function.name fn_args = json.loads(tc.function.arguments) resultado = self._handle_tool_call(fn_name, fn_args) self.conversation_history.append({ “role”: “tool”, “tool_call_id”: tc.id, “content”: result, }) return “(El agente alcanzó el máximo de iteraciones)”
{
“type”: “function”,
“function”: {
“name”: “task_update”,
“description”: “Update the status of a task. Use ‘in_progress’ when starting, ‘completed’ when done.”,
“parameters”: {
“type”: “object”,
“properties”: {
“task_id”: {“type”: “string”, “description”: “The task ID”},
“status”: {“type”: “string”, “enum”: [“in_progress”, “completed”, “failed”]}, “resultado”: {“tipo”: “cadena”, “descripción”: “Resultado o salida de la tarea”}, }, “obligatorio”: [“task_id”, “status”]}, }, }, { “type”: “function”, “function”: { “name”: “inbox_send”, “description”: “Enviar un mensaje a otro agente (p. ej., ‘líder’ o el nombre de un trabajador).”, “parameters”: { “type”: “object”, “properties”: { “to”: {“type”: “string”, “description”: “Nombre del agente destinatario”}, “message”: {“type”: “string”, “description”: “Contenido del mensaje”}, }, “obligatorio”: [“to”, “message”]}, }, }, { “type”: “function”, “function”: { “name”: “inbox_receive”, “description”: “Comprueba y consume todos los mensajes en tu bandeja de entrada.”, “parameters”: { “type”: “object”, “properties”: {}, }, }, }, { “type”: “function”, “function”: { “name”: “task_list”, “description”: “Enumera las tareas asignadas a ti o a todas las tareas del equipo.”, “parámetros”: { “tipo”: “objeto”, “properties”: { “propietario”: {“tipo”: “cadena”, “descripción”: “Filtrar por nombre de propietario (opcional)”}, }, }, }, }, ]clase SwarmAgent: def __init__( self, nombre: str, rol: str, system_prompt: str, task_board: TaskBoard, bandeja de entrada: InboxSystem, registro: TeamRegistry,): self.name = nombre self.role = rol self.system_prompt = system_prompt self.task_board = task_board self.inbox = bandeja de entrada self.registry = registro self.conversation_history: lista[dict] = []
self.inbox.register(nombre) self.registry.register(nombre, rol) def _build_system_prompt(self) -> str: coord_protocol = f””” ## Protocolo de coordinación (autoinyectado: eres el agente ‘{self.name}’) Eres parte de un enjambre de agentes de IA. Tu rol: {self.role} Tu nombre: {self.name} Herramientas disponibles (equivalentes a ClawTeam CLI): – task_list: verifique sus tareas asignadas (como `clawteam task list`) – task_update: actualice el estado de la tarea a in_progress/completed/failed (como `clawteam task update`) – inbox_send: envíe mensajes a otros agentes (como `clawteam inbox send`) – inbox_receive: revise su bandeja de entrada en busca de mensajes (como `clawteam inboxceived`) FLUJO DE TRABAJO: 1. Verifique sus tareas con task_list 2. Marque una tarea como en progreso cuando comienza 3. Haga el trabajo (piense, analice, produzca resultados) 4. Marque la tarea como completada con su resultado 5. Envíe un mensaje de resumen al ‘líder’ cuando haya terminado “”” return self.system_prompt + “\n” + coord_protocol def _handle_tool_call(self, nombre_herramienta: str, args: dict) -> str: if nombre_herramienta == “actualización_tarea”: estado = Estado de la tarea (argumentos)[“status”]) resultado = args.get(“resultado”, “”) self.task_board.update_status(args[“task_id”]estado, resultado) si estado == TaskStatus.COMPLETED: self.registry.increment_completed(self.name) return json.dumps({“ok”: True, “task_id”: args[“task_id”]”nuevo_estado”: argumentos[“status”]}) elif nombre_herramienta == “inbox_send”: self.inbox.send(self.nombre, argumentos[“to”]argumentos[“message”]) devuelve json.dumps({“ok”: Verdadero, “sent_to”: args[“to”]}) elif nombre_herramienta == “inbox_receive”: msgs = self.inbox.receive(self.name) si no es mensaje: return json.dumps({“messages”: []”nota”: “No hay mensajes nuevos”}) return json.dumps({ “mensajes”: [
{“from”: m.sender, “content”: m.content, “time”: m.timestamp}
for m in msgs
]
}) elif nombre_herramienta == “lista_tareas”: propietario = args.get(“propietario”, self.nombre) tareas = self.task_board.get_tasks(propietario=propietario) return json.dumps({“tareas”: [t.to_dict() for t in tasks]}) return json.dumps({“error”: f”Herramienta desconocida: {tool_name}”}) def run(self, user_message: str, max_iterations: int = 6) -> str: self.conversation_history.append({“role”: “user”, “content”: user_message}) para la iteración en el rango (max_iterations): intente: respuesta = client.chat.completions.create(modelo=MODELO, mensajes=[
{“role”: “system”, “content”: self._build_system_prompt()},
*self.conversation_history,
]herramientas=SWARM_TOOLS, tool_choice=”auto”, temperatura=0.4, ) excepto Excepción como e: devolver f”[API Error] {e}” elección = respuesta.opciones[0]
msg = elección.message asistente_msg = {“role”: “asistente”, “contenido”: msg.content o “”} si msg.tool_calls: asistente_msg[“tool_calls”] = [
{
“id”: tc.id,
“type”: “function”,
“function”: {“name”: tc.function.name, “arguments”: tc.function.arguments},
}
for tc in msg.tool_calls
]
self.conversation_history.append(assistant_msg) si no es msg.tool_calls: devuelve msg.content o “(Sin respuesta)” para tc en msg.tool_calls: fn_name = tc.function.name fn_args = json.loads(tc.function.arguments) resultado = self._handle_tool_call(fn_name, fn_args) self.conversation_history.append({ “role”: “tool”, “tool_call_id”: tc.id, “content”: result, }) return “(El agente alcanzó el máximo de iteraciones)”