En este tutorial, profundizamos en nanobot, el marco de agente de IA personal ultraligero de HKUDS que incluye capacidades completas de agente en aproximadamente 4000 líneas de Python. En lugar de simplemente instalarlo y ejecutarlo desde el primer momento, abrimos el capó y recreamos manualmente cada uno de sus subsistemas principales, el bucle del agente, la ejecución de herramientas, la persistencia de la memoria, la carga de habilidades, la gestión de sesiones, la generación de subagentes y la programación cron, para que entendamos exactamente cómo funcionan. Conectamos todo con gpt-4o-mini de OpenAI como nuestro proveedor de LLM, ingresamos nuestra clave API de forma segura a través de la terminal (nunca la exponemos en la salida del cuaderno) y construimos progresivamente desde un único bucle de llamada de herramientas hasta un proceso de investigación de varios pasos que lee y escribe archivos, almacena memorias a largo plazo y delega tareas a trabajadores en segundo plano concurrentes. Al final, no sólo sabemos cómo utilizar nanobots, sino que también sabemos cómo ampliarlos con herramientas personalizadas, habilidades y nuestras propias arquitecturas de agentes.
import sys import os import subprocess def sección(title, emoji="🔹"): """Imprimir un encabezado de sección.""" ancho = 72 print(f"n{'═' * ancho}") print(f" {emoji} {título}") print(f"{'═' * ancho}n") def info(msg): print(f" ℹ️ {msg}") def Success(msg): print(f" ✅ {msg}") def code_block(código): print(f" ┌───────────────────────── ────────────────────────") para la línea en code.strip().split("n"): print(f" │ {línea}") print(f" └───────────────────────── ────────────────────────") sección("PASO 1 · Instalación de nanobot-ai y dependencias", "📦") info("Instalación de nanobot-ai desde PyPI (último estable)…") subprocess.check_call([ sys.executable, "-m", "pip", "install", "-q", "nanobot-ai", "openai", "rich", "httpx" ]) Success("nanobot-ai se instaló correctamente!") import importlib.metadata nanobot_version = importlib.metadata.version("nanobot-ai") print(f" 📌 versión de nanobot-ai: {nanobot_version}") sección("PASO 2 · Entrada segura de la clave API de OpenAI", "🔑") info("Su clave API NO se imprimirá ni almacenará en la salida del cuaderno.") info("Se conserva solo en la memoria para esta sesión.n") intente: desde google.colab importe datos de usuario OPENAI_API_KEY = userdata.get("OPENAI_API_KEY") si no es OPENAI_API_KEY: elevar ValueError("No configurado en secretos de Colab") Success("Clave API cargada de Colab Secrets ('OPENAI_API_KEY').") info("Consejo: puede configurar esto en Colab → 🔑 Panel de secretos en la barra lateral izquierda.") excepto excepción: import getpass OPENAI_API_KEY = getpass.getpass("Ingrese su clave API de OpenAI: ") Success("Clave API capturada de forma segura a través de la entrada del terminal.") os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY import openai client = openai.OpenAI(api_key=OPENAI_API_KEY) try: client.models.list() Success("Clave API OpenAI validada – ¡conexión exitosa!") excepto excepción como e: print(f" ❌ Error en la validación de la clave API: {e}") print(" Reinicie e ingrese una clave válida.") sys.exit(1) sección("PASO 3 · Configuración de nanobot para OpenAI", "⚙️") import json from pathlib import Path NANOBOT_HOME = Path.home() / ".nanobot" NANOBOT_HOME.mkdir(parents=True, exist_ok=True) WORKSPACE = NANOBOT_HOME / "espacio de trabajo" WORKSPACE.mkdir(parents=True, exist_ok=True) (WORKSPACE / "memory").mkdir(parents=True, exist_ok=True) config = { "proveedores": { "openai": { "apiKey": OPENAI_API_KEY } }, "agentes": { "defaults": { "modelo": "openai/gpt-4o-mini", "maxTokens": 4096, "workspace": str(WORKSPACE) } }, "tools": { "restrictToWorkspace": True } } config_path = NANOBOT_HOME / "config.json" config_path.write_text(json.dumps(config, indent=2)) Success(f"Configuración escrita en {config_path}") agentes_md = WORKSPACE / "AGENTS.md" agentes_md.write_text( "# Instrucciones del agentenn" "Eres nanobot 🐈, un asistente personal de IA ultraligero.n" "Eres útil, conciso y utilizas herramientas cuando es necesario.n" "Siempre explica tu razonamiento paso a paso.n" ) soul_md = WORKSPACE / "SOUL.md" soul_md.write_text( "# Personalidadnn" "- Amigable y accesiblen" "- Técnicamente precison" "- Utiliza emoji con moderación para brindar calidezn" ) user_md = WORKSPACE / "USER.md" user_md.write_text( "# Perfil de usuarionn" "- El usuario está explorando el marco de nanobot.n" "- Están interesados en arquitecturas de agentes de IA.n" ) Memory_md = WORKSPACE / "memory" / "MEMORY.md" Memory_md.write_text("# Memoria a largo plazonn_No hay recuerdos almacenados todavía._n") Success("Archivos de arranque del espacio de trabajo creados:") para f en [agents_md, soul_md, user_md, Memory_md]: print(f" 📄 {f.relative_to(NANOBOT_HOME)}") sección ("PASO 4 · Análisis profundo de la arquitectura nanobot", "🏗️") info("""nanobot está organizado en 7 subsistemas en ~4000 líneas de código: ┌───────────────────────────── ─────────────────────────────┐ │ INTERFACES DE USUARIO │ │ CLI · Telegram · WhatsApp · Discord │ └──────────────────┬────────── ─────────────────────────────┘ │ Mensaje entrante / Mensaje saliente ┌──────────────────▼────────── ─────────────────────────────┐ │ BUS DE MENSAJES │ │ publicar_entrada() / publicar_salida() │ └──────────────────┬────────── ─────────────────────────────┘ │ ┌──────────────────▼────────── ─────────────────────────────┐ │ AGENTE LOOP (loop.py) │ │ ┌─────────┐ ┌──────────┐ ┌────────────────────┐ │ │ │ Contexto │→ │ LLM │→ │ Ejecución de herramientas │ │ │ │ Constructor │ │ Llamar │ │ (si la herramienta_llama) │ │ │ └─────────┘ └──────────┘ └────────┬───────────┘ │ │ ▲ │ bucle hacia atrás │ │ │ ◄───────────────────┘ hasta terminar │ │ ┌────┴────┐ ┌──────────┐ ┌────────────────────┐ │ │ │ Memoria │ │ Habilidades │ │ Subagente Mgr │ │ │ │ Tienda │ │ Cargador │ │ (tareas de generación) │ │ │ └─────────┘ └──────────┘ └────────────────────┘ │ └───────────────────────────── ─────────────────────────────┘ │ ┌──────────────────▼────────── ─────────────────────────────┐ │ CAPA DE PROVEEDOR LLM │ │ OpenAI · Anthropic · OpenRouter · DeepSeek · … │ └───────────────────────────── ──────────────────────────────┘ El bucle del agente se repite hasta 40 veces (configurable): 1. ContextBuilder ensambla el indicador del sistema + memoria + habilidades + historial 2. LLM se llama con definiciones de herramientas 3. Si la respuesta tiene llamadas a herramientas → ejecutar herramientas, agregar resultados, bucle 4. Si la respuesta es texto sin formato → devolver como respuesta final """)
Configuramos la base completa del tutorial importando los módulos necesarios, definiendo funciones auxiliares para la visualización de secciones limpias e instalando las dependencias de nanobot dentro de Google Colab. Luego cargamos y validamos de forma segura la clave API de OpenAI para que el resto del cuaderno pueda interactuar con el modelo sin exponer las credenciales en la salida del cuaderno. Después de eso, configuramos el espacio de trabajo del nanobot y creamos los archivos de arranque principales, como AGENTS.md y SOUL.md, USER.md y MEMORY.md, y estudiamos la arquitectura de alto nivel para comprender cómo está organizado el marco antes de pasar a la implementación.
sección("PASO 5 · El bucle del agente: concepto central en acción", "🔄") info("Recrearemos manualmente el patrón del bucle del agente de nanobot usando OpenAI.") info("Esto es exactamente lo que loop.py hace internamente.n") import json as _json import datetime TOOLS = [ { "type": "function", "function": { "name": "get_current_time", "description": "Obtener la fecha y hora actuales.", "parameters": {"tipo": "objeto", "propiedades": {}, "requerido":[]} } }, { "type": "function", "function": { "name": "calcular", "description": "Evaluar una expresión matemática.", "parameters": { "type": "object", "properties": { "expression": { "type": "string", "description": "Expresión matemática a evaluar, por ejemplo, '2**10 + 42'" } }, "required": ["expression"] } } }, { "type": "function", "function": { "name": "read_file", "description": "Leer el contenido de un archivo en el espacio de trabajo.", "parameters": { "type": "object", "properties": { "path": { "type": "string", "description": "Ruta relativa del archivo dentro del workspace" } }, "required": ["path"] } } }, { "type": "function", "function": { "name": "write_file", "description": "Escribir contenido en un archivo en el espacio de trabajo.", "parameters": { "type": "object", "properties": { "path": {"type": "string", "description": "Ruta relativa al archivo"}, "content": {"type": "string", "description": "Contenido para escribir"} }, "required": ["path", "content"] } } }, { "type": "function", "function": { "name": "save_memory", "description": "Guardar un hecho en la memoria a largo plazo del agente.", "parameters": { "type": "object", "properties": { "fact": {"type": "string", "description": "El hecho a recordar"} }, "required": ["fact"] } } } ] def ejecutar_tool(nombre: str, argumentos: dict) -> str: """Ejecutar una llamada a una herramienta: refleja ToolRegistry.execute() del nanobot.""" if name == "get_current_time": elif name == "calculate": expr = arguments.get("expression", "") try: result = eval(expr, {"__builtins__": {}}, {"abs": abs, "round": round, "min": min, "max": max}) return str(result) excepto excepción como e: return f"Error: {e}" elif name == "read_file": fpath = WORKSPACE / arguments.get("path", "") if fpath.exists(): return fpath.read_text()[:4000] return f"Error: Archivo no encontrado — {arguments.get('path')}" elif name == "write_file": fpath = WORKSPACE / arguments.get("path", "") fpath.parent.mkdir(parents=True, exist_ok=True) fpath.write_text(arguments.get("content", "")) return f"Escribió con éxito {len(arguments.get('content', ''))} caracteres en {arguments.get('path')}" elif name == "save_memory": fact = arguments.get("fact", "") mem_file = WORKSPACE / "memoria" / "MEMORIA.md" existente = mem_file.read_text() marca de tiempo = datetime.datetime.now().strftime("%Y-%m-%d %H:%M") mem_file.write_text(existente + f"n- [{marca de tiempo}] {fact}n") return f"Memoria guardada: {fact}" return f"Herramienta desconocida: {nombre}" def agent_loop(user_message: str, max_iterations: int = 10, verbose: bool = True): """ Recrea la lógica AgentLoop._process_message() del nanobot. El bucle: 1. Construir contexto (indicador del sistema + archivos de arranque + memoria) 2. Llamar a LLM con herramientas 3. Si tool_calls → ejecutar → agregar resultados → bucle 4. Si respuesta de texto → devolver respuesta final """ system_parts =[]para md_file en ["AGENTS.md", "SOUL.md", "USER.md"]: fpath = WORKSPACE / md_file if fpath.exists(): system_parts.append(fpath.read_text()) mem_file = WORKSPACE / "memory" / "MEMORY.md" if mem_file.exists(): system_parts.append(f"n## Su Memorian{mem_file.read_text()}") system_prompt = "nn".join(system_parts) mensajes = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message} ] si es detallado: print(f" 📨 Usuario: {user_message}") print(f" 🧠 Mensaje del sistema: {len(system_prompt)} caracteres "f"(de {len(system_parts)} archivos de arranque)") print() para iteración en el rango(1, max_iterations + 1): si es detallado: print(f" ── Iteración {iteración}/{max_iterations} ──") respuesta = client.chat.completions.create( model="gpt-4o-mini", mensajes=mensajes, herramientas=HERRAMIENTAS, tool_choice="auto", max_tokens=2048 ) elección = respuesta.opciones[0]mensaje = elección.message si mensaje.tool_calls: si detallado: print(f" 🔧 LLM solicitó {len(message.tool_calls)} llamada(s) a herramientas:") mensajes.append(message.model_dump()) para tc en message.tool_calls: fname = tc.function.name args = _json.loads(tc.function.arguments) if tc.function.arguments else {} si es detallado: print(f" → {fname}({_json.dumps(args, sure_ascii=False)[:80]})") resultado = ejecutar_tool(fname, args) si es detallado: print(f" ← {resultado[:100]}{'…' si len(resultado) > 100 else ''}") mensajes.append({ "role": "tool", "tool_call_id": tc.id, "content": resultado }) if detallado: print() else: final = message.content o "" if detallado: print(f" 💬 Agente: {final}n") return final return "⚠️ Se alcanzaron el máximo de iteraciones sin una respuesta final". print("─" * 60) print(" DEMO 1: Cálculo basado en el tiempo con encadenamiento de herramientas") print("─" * 60) result1 = agent_loop( "¿Cuál es la hora actual? Además, calcula 2^20 + 42 para mí." ) print("─" * 60) print(" DEMO 2: Creación de archivos + almacenamiento en memoria") print("─" * 60) resultado2 = agent_loop( "Escribe un haiku sobre agentes de IA en un archivo llamado 'haiku.txt'. " "Entonces recuerda que disfruto la poesía sobre tecnología." )
Recreamos manualmente el corazón del nanobot definiendo los esquemas de las herramientas, implementando su lógica de ejecución y construyendo el bucle de agente iterativo que conecta el LLM con las herramientas. Ensamblamos el mensaje a partir de los archivos del espacio de trabajo y la memoria, enviamos la conversación al modelo, detectamos llamadas a herramientas, las ejecutamos, agregamos los resultados nuevamente a la conversación y seguimos repitiendo hasta que el modelo devuelve una respuesta final. Luego probamos este mecanismo con ejemplos prácticos que involucran búsquedas de tiempo, cálculos, escritura de archivos y ahorro de memoria, para que podamos ver el ciclo operar exactamente como el flujo interno del nanobot.
sección("PASO 6 · Sistema de memoria — Memoria de agente persistente", "🧠") info("""El sistema de memoria del nanobot (memory.py) utiliza dos mecanismos de almacenamiento: 1. MEMORY.md: datos a largo plazo (siempre cargados en contexto) 2. AAAA-MM-DD.md: entradas del diario (cargadas para los últimos días) La consolidación de la memoria se ejecuta periódicamente para resumir y comprimir las entradas antiguas, manteniendo la ventana de contexto manejable. """) mem_content = (ESPACIO DE TRABAJO / "memoria" / "MEMORIA.md").read_text() print(" 📂 Contenido actual de MEMORIA.md:") print(" ┌─────────────────────── ──────────────────────") para la línea en mem_content.strip().split("n"): print(f" │ {línea}") print(" └──────────────────────── ─────────────────────n") today = datetime.datetime.now().strftime("%Y-%m-%d") daily_file = WORKSPACE / "memory" / f"{today}.md" daily_file.write_text( f"# Registro diario — {today}nn" "- El usuario ejecutó el tutorial avanzado del nanobotn" "- Explora el bucle del agente, las herramientas y la memorian" "- Creó un haiku sobre los agentes de IAn" ) Success(f"Diario creado: memoria/{hoy}.md") print("n 📁 Contenido del espacio de trabajo:") para el elemento en ordenado(WORKSPACE.rglob("*")): if item.is_file(): rel = item.relative_to(WORKSPACE) size = item.stat().st_size print(f" {'📄' if item.suffix == '.md' else '📝'} {rel} ({size} bytes)") sección ("PASO 7 · Sistema de habilidades: ampliación de las capacidades del agente", "🎯") info("""SkillsLoader (skills.py) del nanobot lee los archivos Markdown del directorio skills/. Cada habilidad tiene: – Un nombre y una descripción (para que el LLM decida cuándo usarlo) – Instrucciones que el LLM sigue cuando se activa la habilidad – Algunas habilidades están 'siempre cargadas'; otras se cargan a pedido Creemos una habilidad personalizada y veamos cómo la usa el agente. """) skills_dir = WORKSPACE / "skills" skills_dir.mkdir(exist_ok=True) data_skill = skills_dir / "data_analyst.md" data_skill.write_text("""# Habilidad de analista de datos ## Descripción Analizar datos, calcular estadísticas y proporcionar información a partir de números. ## Instrucciones Cuando se le solicite analizar datos: 1. Identificar los datos. tipo y estructura 2. Calcular estadísticas relevantes (media, mediana, rango, desarrollo estándar) 3. Buscar patrones y valores atípicos 4. Presentar los hallazgos en un formato claro y estructurado 5. Sugerir preguntas de seguimiento ## Siempre disponible false """) review_skill = skills_dir / "code_reviewer.md" review_skill.write_text("""# Habilidad del revisor de código ## Descripción Revisar el código para detectar errores, problemas de seguridad y mejores prácticas. ## Instrucciones Al revisar el código: 1. Verificar errores comunes y lógicos 2. Identificar vulnerabilidades de seguridad 3. Sugerir mejoras de rendimiento 4. Evaluar el estilo y la legibilidad del código 5. Calificar la calidad del código en una escala del 1 al 10 ## Siempre disponible true """) Success("Habilidades personalizadas creadas:") for f in skills_dir.iterdir(): print(f" 🎯 {f.name}") print("n 🧪 Pruebas interacción del agente con reconocimiento de habilidades:") print(" " + "─" * 56) skills_context = "nn## Skills disponiblesn" for skills_file in skills_dir.glob("*.md"): content = skills_file.read_text() skills_context += f"n### {skill_file.stem}n{content}n" result3 = agent_loop( "Revise este código Python para problemas:nn" "“`pythonn" "def get_user(id):n" " consulta = f'SELECT * FROM usuarios DONDE id = {id}'n" " resultado = db.execute(query)n" " return resultadon" "“`" )
Pasamos al sistema de memoria persistente inspeccionando el archivo de memoria a largo plazo, creando una entrada diaria y revisando cómo evoluciona el espacio de trabajo después de interacciones anteriores. Luego ampliamos el agente con un sistema de habilidades mediante la creación de archivos de habilidades basados en rebajas que describen comportamientos especializados, como análisis de datos y revisión de código. Finalmente, simulamos cómo funcionan las indicaciones basadas en habilidades exponiendo estas habilidades al agente y pidiéndole que revise una función de Python, lo que nos ayuda a ver cómo se puede guiar al nanobot a través de descripciones de capacidades modulares.
sección("PASO 8 · Creación de herramientas personalizadas: Ampliación del agente", "🔧") info("""El sistema de herramientas de nanobot utiliza un ToolRegistry con una interfaz simple. Cada herramienta necesita: – Un nombre y descripción – Un esquema JSON para parámetros – Un método de ejecución() Creemos herramientas personalizadas y conéctelas a nuestro bucle de agente. """) import random CUSTOM_TOOLS = [ { "type": "function", "function": { "name": "roll_dice", "description": "Lanza uno o más dados con un número determinado de caras.", "parameters": { "type": "object", "properties": { "num_dice": {"type": "integer", "description": "Número de dados a tirar", "default": 1}, "sides": {"type": "integer", "description": "Número de caras por dado", "default": 6} }, "required":[]} } }, { "type": "function", "function": { "name": "text_stats", "description": "Calcular estadísticas sobre un texto: recuento de palabras, recuento de caracteres, recuento de oraciones, tiempo de lectura.", "parameters": { "type": "object", "properties": { "text": {"type": "string", "description": "El texto a analizar"} }, "required": ["text"] } } }, { "type": "función", "función": { "nombre": "generar_contraseña", "descripción": "Generar una contraseña segura aleatoria.", "parámetros": { "tipo": "objeto", "properties": { "longitud": {"tipo": "entero", "descripción": "Longitud de la contraseña", "predeterminado": 16} }, "obligatorio":[]} } } ] _original_execute = ejecutar_herramienta def ejecutar_herramienta_extendida(nombre: str, argumentos: dict) -> str: if nombre == "roll_dice": n = argumentos.get("num_dice", 1) s = argumentos.get("lados", 6) rolls = [random.randint(1, s) for _ in range(n)] return f"Rolled {n}d{s}: {rollos} (total: {suma(rollos)})" elif nombre == "text_stats": texto = argumentos.get("text", "") palabras = len(text.split()) caracteres = len(texto) oraciones = text.count('.') + text.count('!') + text.count('?') tiempo_lectura = round(palabras / 200, 1) return _json.dumps({ "palabras": palabras, "caracteres": caracteres, "sentencias": max(sentencias, 1), "tiempo_lectura_minutos": tiempo_lectura }) nombre elif == "generar_contraseña": cadena de importación longitud = argumentos.get("longitud", 16) caracteres = cadena.ascii_letters + cadena.dígitos + "!@#$%^&*" pwd = ''.join(random.choice(chars) for _ in range(length)) return f"Contraseña generada ({length} caracteres): {pwd}" return _original_execute(nombre, argumentos) ejecutar_tool = ejecutar_tool_extended ALL_TOOLS = TOOLS + CUSTOM_TOOLS def agent_loop_v2(user_message: str, max_iterations: int = 10, detallado: bool = Verdadero): """Bucle de agente con herramientas personalizadas extendidas.""" system_parts =[]para md_file en ["AGENTS.md", "SOUL.md", "USER.md"]: fpath = WORKSPACE / md_file if fpath.exists(): system_parts.append(fpath.read_text()) mem_file = WORKSPACE / "memory" / "MEMORY.md" if mem_file.exists(): system_parts.append(f"n## Su Memorian{mem_file.read_text()}") system_prompt = "nn".join(system_parts) mensajes = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message} ] if detallado: print(f" 📨 Usuario: {user_message}") print() para iteración en rango(1, max_iterations + 1): if detallado: print(f" ── Iteración {iteración}/{max_iterations} ──") respuesta = client.chat.completions.create( model="gpt-4o-mini", mensajes=mensajes, herramientas=ALL_TOOLS, tool_choice="auto", max_tokens=2048 ) elección = respuesta.opciones[0]mensaje = elección.message si mensaje.tool_calls: si detallado: print(f" 🔧 {len(message.tool_calls)} llamada(s) a herramienta):") mensajes.append(message.model_dump()) para tc en message.tool_calls: fname = tc.function.name args = _json.loads(tc.function.arguments) if tc.function.arguments else {} if detallado: print(f" → {fname}({_json.dumps(args, sure_ascii=False)[:80]})") resultado = ejecutar_tool(fname, args) if detallado: print(f" ← {resultado[:120]}{'…' if len(resultado) > 120 else ''}") mensajes.append({ "role": "tool", "tool_call_id": tc.id, "content": resultado }) si es detallado: print() else: final = message.content o "" si es detallado: print(f" 💬 Agente: {final}n") return final return "⚠️ Se alcanzó el máximo de iteraciones". print("─" * 60) print(" DEMO 3: Herramientas personalizadas en acción") print("─" * 60) result4 = agent_loop_v2( "Tira 3 dados de seis caras por mí, luego genera una contraseña de 20 caracteres, " "y finalmente analiza las estadísticas de texto de esta oración: " ) sección("PASO 9 · Conversación de varios turnos – Gestión de sesiones", "💬") info("""SessionManager de nanobot (session/manager.py) mantiene el historial de conversaciones por clave de sesión (formato: 'canal:chat_id'). El historial se almacena en archivos JSON y se carga en contexto para cada mensaje nuevo. Simulemos una conversación de varios turnos con estado persistente. """)
Ampliamos las capacidades del agente definiendo nuevas herramientas personalizadas, como lanzamiento de dados, estadísticas de texto y generación de contraseñas, y luego las conectamos al proceso de ejecución de herramientas. Actualizamos el ejecutor, fusionamos las definiciones de herramientas integradas y personalizadas y creamos una segunda versión del bucle del agente que puede razonar sobre este conjunto más amplio de capacidades. Luego ejecutamos una tarea de demostración que obliga al modelo a encadenar múltiples invocaciones de herramientas, demostrando lo fácil que es extender el nanobot con nuestras propias funciones manteniendo el mismo patrón de interacción general.
class SimpleSessionManager: """ Recreación mínima del SessionManager de nanobot. Almacena el historial de conversaciones y proporciona continuidad del contexto. """ def __init__(self, workspace: Path): self.workspace = workspace self.sessions: dict[str, list[dict]] = {} def get_history(self, session_key: str) -> list[dict]: return self.sessions.get(session_key,[]) def add_turn(self, clave_sesión: cadena, rol: cadena, contenido: cadena): si clave_sesión no está en self.sessions: self.sessions[clave_sesión] =[]self.sessions[clave_sesión].append({"role": rol, "content": contenido}) def save(self, clave_sesión: str): fpath = self.workspace / f"session_{session_key.replace(':', '_')}.json" fpath.write_text(_json.dumps(self.sessions.get(session_key,[]), indent=2)) def load(self, session_key: str): fpath = self.workspace / f"session_{session_key.replace(':', '_')}.json" if fpath.exists(): self.sessions[session_key] = _json.loads(fpath.read_text()) session_mgr = SimpleSessionManager(WORKSPACE) SESSION_KEY = "cli:tutorial_user" def chat(user_message: str, verbose: bool = True): """Chat multiturno con persistencia de sesión.""" session_mgr.add_turn(SESSION_KEY, "user", user_message) system_parts =[]para md_file en ["AGENTS.md", "SOUL.md"]: fpath = WORKSPACE / md_file if fpath.exists(): system_parts.append(fpath.read_text()) system_prompt = "nn".join(system_parts) historial = session_mgr.get_history(SESSION_KEY) mensajes = [{"role": "system", "content": system_prompt}] + historial si es detallado: print(f" 👤 Tú: {user_message}") print(f" (historial de conversaciones: {len(history)} mensajes)") respuesta = client.chat.completions.create( model="gpt-4o-mini", mensajes=messages, max_tokens=1024 ) respuesta = respuesta.choices[0].message.content o "" session_mgr.add_turn(SESSION_KEY, "assistant", respuesta) session_mgr.save(SESSION_KEY) si es detallado: print(f" 🐈 nanobot: {respuesta}n") return respuesta print("─" * 60) print(" DEMO 4: Conversación multiturno con memoria") print("─" * 60) chat("¡Hola! Mi mi nombre es Alex y estoy creando un agente de IA.") chat("¿Cómo me llamo? ¿Y en qué estoy trabajando?") chat("¿Puede sugerir 3 funciones que debería agregar a mi agente?") Success("¡La sesión persistió con el historial de conversaciones completo!") session_file = WORKSPACE / f"session_{SESSION_KEY.replace(':', '_')}.json" session_data = _json.loads(session_file.read_text()) print(f" 📄 Archivo de sesión: {session_file.name} ({len(session_data)} mensajes)") sección ("PASO 10 · Generación de subagente: delegación de tareas en segundo plano", "🚀") info("""SubagentManager de nanobot (agent/subagent.py) permite al agente principal delegar tareas a trabajadores independientes en segundo plano. Cada subagente: – Obtiene su propio registro de herramientas (sin SpawnTool) para evitar la recursión) – Ejecuta hasta 15 iteraciones de forma independiente – Informa los resultados a través de MessageBus Simulemos este patrón con tareas concurrentes """) import asyncio import uuid async def run_subagent(task_id: str, goal: str, verbose: bool = True): """ Simula SubagentManager._run_subagent() de nanobot Ejecuta un bucle LLM independiente para un objetivo específico. """ si es detallado: print(f" 🔹 Subagente [{task_id[:8]}] iniciado: {goal[:60]}") respuesta = client.chat.completions.create( model="gpt-4o-mini", mensajes=[ {"role": "system", "content": "Eres un asistente de investigación enfocado". "Completa la tarea asignada de manera concisa en 2-3 oraciones."}, {"role": "user", "contenido": objetivo}], max_tokens=256) resultado = respuesta.opciones[0].message.content o "" si es detallado: print(f" ✅ Subagente [{task_id[:8]}] hecho: {resultado[:80]}…") return {"task_id": task_id, "goal": objetivo, "resultado": resultado} async def spawn_subagents(goals: list[str]): """Generar múltiples subagentes simultáneamente – mirrors SubagentManager.spawn().""" tareas =[]para objetivo en objetivos: task_id = str(uuid.uuid4()) tareas.append(run_subagent(task_id, objetivo)) print(f"n 🚀 Generando {len(tasks)} subagentes simultáneamente…n") resultados = await asyncio.gather(*tasks) devuelve resultados objetivos = [ "¿Cuáles son los 3 componentes clave de una arquitectura de agente ReAct?", "Explique la diferencia entre llamadas de herramientas y llamadas de funciones en LLM.", "¿Qué es MCP (Protocolo de contexto modelo) y por qué es importante para los agentes de IA?", ] intente: loop = asyncio.get_running_loop() import nest_asyncio nest_asyncio.apply() subagent_results = asyncio.get_event_loop().run_until_complete(spawn_subagents(goals)) excepto RuntimeError: subagent_results = asyncio.run(spawn_subagents(goals)) excepto ModuleNotFoundError: print(" ℹ️ Ejecutando subagents secuencialmente (instalar nest_asyncio para async)…n") subagent_results =[]para objetivo en objetivos: task_id = str(uuid.uuid4()) respuesta = client.chat.completions.create( model="gpt-4o-mini", mensajes=[ {"role": "system", "content": "Completa la tarea de forma concisa en 2-3 oraciones."}, {"role": "user", "content": goal} ], max_tokens=256 ) r = respuesta.choices[0].message.content o "" print(f" ✅ Subagente [{task_id[:8]}] hecho: {r[:80]}…") subagent_results.append({"task_id": task_id, "goal": goal, "result": r}) print(f"n 📋 ¡Todos los {len(subagent_results)} resultados del subagent recopilados!") para i, r en enumerar(subagent_results, 1): print(f"n ── Resultado {i} ──") print(f" Objetivo: {r['meta'][:60]}") print(f" Respuesta: {r['resultado'][:200]}")
Simulamos la gestión de conversaciones en varios turnos mediante la creación de un administrador de sesiones liviano que almacena, recupera y conserva el historial de conversaciones en todos los turnos. Usamos ese historial para mantener la continuidad en el chat, lo que permite al agente recordar detalles anteriores de la interacción y responder de manera más coherente y detallada. Después de eso, modelamos la generación de subagentes lanzando tareas en segundo plano simultáneas, cada una de las cuales maneja un objetivo enfocado, lo que nos ayuda a comprender cómo el nanobot puede delegar el trabajo paralelo a agentes trabajadores independientes.
sección("PASO 11 · Tareas programadas: el patrón Cron", "⏰") info("""CronService (cron/service.py) de nanobot usa APScheduler para desencadenar acciones del agente según una programación. Cuando se activa un trabajo, crea un InboundMessage y lo publica en MessageBus. Demostremos el patrón con un programador simulado. """) de la clase SimpleCronJob de importación de fecha y hora: """Refleja la estructura del trabajo cron del nanobot.""" def __init__(self, nombre: str, mensaje: str, intervalo_segundos: int): self.id = str(uuid.uuid4())[:8] self.name = nombre self.message = mensaje self.interval = intervalo_segundos self.enabled = Verdadero self.last_run = Ninguno self.next_run = datetime.datetime.now() + timedelta(segundos=interval_segundos) jobs = [ SimpleCronJob("morning_briefing", "Dame una breve actualización de estado por la mañana.", 86400), SimpleCronJob("memory_cleanup", "Revisar y consolidar mis recuerdos.", 43200), SimpleCronJob("health_check", "Ejecutar una verificación del estado del sistema.", 3600), ] print(" 📋 Registrado Trabajos cron:") imprimir(" ┌────────┬────────────────────┬── ────────┬──────────────────────┐") print(" │ ID │ Nombre │ Intervalo │ Próxima ejecución │") print(" ├────────┼────────────────────┼── ────────┼──────────────────────┤") para trabajo en trabajos: intervalo_str = f"{job.interval // 3600}h" if trabajo.interval >= 3600 else f"{job.interval}s" print(f" │ {job.id} │ {job.name:<18} │ {interval_str:>8} │ {job.next_run.strftime('%Y-%m-%d %H:%M')} │") imprimir(" └────────┴────────────────────┴── ────────┴──────────────────────┘") print(f"n ⏰ Simulando el disparador cron para '{trabajos[2].nombre}'…") cron_result = agent_loop_v2(trabajos[2].message, verbose=True) sección("PASO 12 · Canalización completa del agente: demostración de un extremo a otro", "🎬") info("""Ahora ejecutemos una tarea compleja de varios pasos que ejercita la canalización completa del nanobot: creación de contexto → uso de herramientas → memoria → E/S de archivos. """) print("─" * 60) print(" DEMO 5: Tarea de investigación compleja de varios pasos") print("─" * 60) complex_result = agent_loop_v2( "Necesito que me ayudes con un pequeño proyecto:n" "1. Primero, verifica la hora actualn" "2. Escribe un breve plan de proyecto en 'project_plan.txt' sobre " "crear un asistente personal de IA (3-4 viñetas)n" "3. Recuerda que mi proyecto actual es 'crear un asistente personal de IA'n" "4. Vuelve a leer el archivo del plan del proyecto para confirmar que se guardó correctamenten" "Luego resuma todo lo que hizo.", max_iterations=15 ) sección("PASO 13 · Resumen final del espacio de trabajo", "📊") print(" 📁 Estado completo del espacio de trabajo después del tutorial:n") total_files = 0 total_bytes = 0 para el elemento en sorted(WORKSPACE.rglob("*")): if item.is_file(): rel = item.relative_to(WORKSPACE) size = item.stat().st_size total_files += 1 total_bytes += tamaño icono = {"md": "📄", "txt": "📝", "json": "📋"}.get(item.suffix.lstrip("."), "📎") print(f" {icon} {rel} ({size:,} bytes)") print(f"n ── Resumen ──") print(f" Total de archivos: {total_files}") print(f" Tamaño total: {total_bytes:,} bytes") print(f" Configuración: {config_path}") print(f" Espacio de trabajo: {WORKSPACE}") print("n 🧠 Estado final de la memoria:") mem_content = (WORKSPACE / "memory" / "MEMORY.md").read_text() print(" ┌─────────────────────── ──────────────────────") para la línea en mem_content.strip().split("n"): print(f" │ {línea}") print(" └─────────────────────── ──────────────────────") sección("COMPLETO · ¿Qué sigue?", "🎉") print(""" ¡Has explorado los aspectos internos principales de nanobot! Esto es lo que puedes probar a continuación: 🔹 Ejecute el agente CLI real: nanobot onboard && nanobot agent 🔹 Conéctese a Telegram: agregue un token de bot a config.json y ejecute `nanobot gateway` 🔹 Habilite la búsqueda web: agregue una clave de API de Brave Search en tools.web.search.apiKey 🔹 Pruebe la integración MCP: nanobot admite servidores Model Context Protocol para herramientas externas 🔹 Explore la fuente (~4K líneas): https://github.com/HKUDS/nanobot 🔹 Archivos clave para leer: • agent/loop.py: el bucle de iteración del agente • agent/context.py: proceso de ensamblaje rápido • agent/memory.py: sistema de memoria persistente • agent/tools/: herramienta integrada implementaciones • agente/subagent.py — Delegación de tareas en segundo plano """)
Demostramos el patrón de programación estilo cron definiendo trabajos programados simples, enumerando sus intervalos y tiempos de próxima ejecución, y simulando la activación de una tarea de agente automatizada. Luego ejecutamos un ejemplo más amplio de un extremo a otro que combina la creación de contexto, el uso de herramientas, las actualizaciones de memoria y las operaciones de archivos en un único flujo de trabajo de varios pasos, de modo que podamos ver el proceso completo trabajando en conjunto en una tarea realista. Al final, inspeccionamos el estado final del espacio de trabajo, revisamos la memoria almacenada y cerramos el tutorial con los siguientes pasos claros que conectan esta implementación del cuaderno con el proyecto real de nanobot y su código fuente.
En conclusión, recorrimos cada capa importante de la arquitectura del nanobot, desde el bucle iterativo de la herramienta LLM en su núcleo hasta el administrador de sesión que le brinda a nuestro agente memoria conversacional en todos los turnos. Creamos cinco herramientas integradas, tres herramientas personalizadas, dos habilidades, una capa de persistencia de sesión, un generador de subagente y un simulador de cron, todo ello manteniendo todo en un único script ejecutable. Lo que destaca es cómo el nanobot demuestra que un marco de agente de nivel de producción no necesita cientos de miles de líneas de código; Los patrones que implementamos aquí, ensamblaje de contexto, envío de herramientas, consolidación de memoria y delegación de tareas en segundo plano, son los mismos patrones que impulsan sistemas mucho más grandes, pero reducidos a su esencia. Ahora tenemos un modelo mental funcional de componentes internos de IA agente y una base de código lo suficientemente pequeña como para leerla de una sola vez, lo que hace que el nanobot sea una opción ideal para cualquiera que busque construir, personalizar o investigar agentes de IA desde cero.
Consulte los códigos completos aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 120.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.
Michal Sutter es un profesional de la ciencia de datos con una Maestría en Ciencia de Datos de la Universidad de Padua. Con una base sólida en análisis estadístico, aprendizaje automático e ingeniería de datos, Michal se destaca en transformar conjuntos de datos complejos en conocimientos prácticos.