importar subproceso, sys def _pip_install(*pkgs): intente: subprocess.run([sys.executable, “-m”, “pip”, “install”, “-q”, *pkgs]check=True) excepto Excepción como e: print(f”(la instalación de pip se omitió/falló para {pkgs}: {e})”) _HAVE_OPENAI = Falso intento: importar openai _HAVE_OPENAI = Verdadero excepto Excepción: _pip_install(“openai>=1.0.0”) intento: importar openai _HAVE_OPENAI = Verdadero excepto Excepción: _HAVE_OPENAI = Falso intento: importar nest_asyncio nest_asyncio.apply() excepto Excepción: intente: _pip_install(“nest_asyncio”) import nest_asyncio nest_asyncio.apply() excepto Excepción: pasar importar os importar re importar json importar tiempo importar matemáticas importar asyncio importar inspeccionar importar ajuste de texto importar contextlib importar io desde clases de datos importar clase de datos, campo desde escribir importar Cualquiera, Invocable, Opcional, Aguardable, get_type_hints def banner(título: str) -> Ninguno: línea = “═” * 78 print(f”\n{line}\n {title}\n{line}”) @dataclass class ToolCall: “””Una solicitud normalizada del modelo para ejecutar una herramienta.””” id: str nombre: str argumentos: dict @dataclass class Uso: Prompt_tokens: int = 0 complete_tokens: int = 0 @property def total(self) -> int: return self.prompt_tokens + self.completion_tokens @dataclass clase LLMResponse: “””La forma única que cada proveedor debe devolver.””” contenido: Opcional[str]
llamadas_herramientas: lista[ToolCall] = campo(default_factory=lista) fin_reason: str = “detener” uso: Uso = campo(default_factory=Uso) clase Proveedor: “””Clase base. Un proveedor convierte (mensajes, herramientas) en una respuesta LLMR.””” nombre = “base” async def complete(self, mensajes: lista[dict]herramientas: lista[dict]) -> LLMResponse: elevar la clase NotImplementedError OpenAICompatibleProvider(Provider): “”” Funciona con OpenAI y todas las puertas de enlace compatibles con OpenAI (OpenRouter, DeepSeek, Together, vLLM, LM Studio, Ollama’s /v1, …). Esto refleja cómo nanobot habla con la mayoría de los proveedores bajo el capó. “”” name = “openai-compatible” def __init__(self, api_key: cadena, modelo: cadena, base_url: opcional[str] = Ninguno): de openai import AsyncOpenAI self.model = model self.client = AsyncOpenAI(api_key=api_key, base_url=base_url) async def complete(self, mensajes: lista[dict]herramientas: lista[dict]) -> LLMRespuesta: kwargs: dict[str, Any] = {“modelo”: self.model, “mensajes”: mensajes} si herramientas: kwargs[“tools”] = herramientas kwargs[“tool_choice”] = “auto” resp = await self.client.chat.completions.create(**kwargs) elección = resp.choices[0]
msg = elección.llamadas de mensaje: lista[ToolCall] = []
para tc en (msg.tool_calls o []): intente: args = json.loads(tc.function.arguments o “{}”) excepto json.JSONDecodeError: args = {“_raw”: tc.function.arguments} call.append(ToolCall(id=tc.id, name=tc.function.name, arguments=args)) uso = Uso( Prompt_tokens=getattr(resp.usage, “prompt_tokens”, 0) o 0, complete_tokens=getattr(resp.usage, “completion_tokens”, 0) o 0, ) return LLMResponse( content=msg.content, tool_calls=calls, Finish_reason=choice.finish_reason o “stop”, use=usage, ) clase MockProvider(Provider): “”” Un “LLM” determinista y basado en reglas, por lo que esto Todo el tutorial se ejecuta SIN clave API y SIN red, lo que le permite observar el bucle del agente, las llamadas a la herramienta y el funcionamiento de la memoria. Imita lo ÚNICO que importa para el bucle: decidir emitir una llamada a la herramienta (en la forma normalizada exacta que tendría un modelo real) y luego, una vez que se obtienen los resultados de la herramienta, producir una respuesta final en lenguaje natural. str = “mock-1”): self.model = modelo @staticmethod def _last_user_text(mensajes: lista[dict]) -> str: para m en reverso (mensajes): if m.get(“role”) == “usuario”: c = m.get(“content”) devuelve c si isinstance(c, str) else json.dumps(c) devuelve “” @staticmethod def _already_ceived(mensajes: lista[dict]nombre_herramienta: str) -> bool: para m en mensajes: si m.get(“role”) == “asistente” y m.get(“tool_calls”): para tc en m[“tool_calls”]: si tc[“function”][“name”] == nombre_herramienta: return True return False @staticmethod def _extract_math(text: str) -> str: “””Saque el primer fragmento matemático de una oración (ayudante solo simulado).””” t = re.sub(r”¿raíces cuadradas? of (\d+(?:\.\d+)?)”, r”sqrt(\1)”, text) t = t.replace(“^”, “**”) patrón = (r”(?:sqrt\(\d+(?:\.\d+)?\)|\d+(?:\.\d+)?)” r”(?:\s*(?:\*\*|[\+\-\*\/])\s*(?:sqrt\(\d+(?:\.\d+)?\)|\d+(?:\.\d+)?))*”) m = re.search(pattern, t) return m.group(0).strip() si m más t.strip() @staticmethod def _scan_memory(messages: list[dict]) -> tupla[Optional[str]Opcional[str]]: “””Leer datos simples de turnos de USUARIO anteriores: demuestra que la memoria de la sesión en realidad se está alimentando al modelo (conveniencia solo simulada).””” name = love = Ninguno para m en mensajes: if m.get(“role”) == “user” and isinstance(m.get(“content”), str): tx = m[“content”].lower() nm = re.search(r”mi nombre es (\w+)”, tx) if nm: nombre = nm.group(1).title() lv = re.search(r”i (?:love|like) (\w+)”, tx) if lv: love = lv.group(1).title() return nombre, love async def complete(self, mensajes: lista[dict]herramientas: lista[dict]) -> LLMResponse: await asyncio.sleep(0) usuario = self._last_user_text(messages).lower() nombres_herramientas = {t[“function”][“name”] para t en herramientas} uso = Uso(prompt_tokens=sum(len(str(m)) para m en mensajes) // 4, complete_tokens=12) def call(name, args): return LLMResponse( content=Ninguno, tool_calls=[ToolCall(id=f”call_{name}_{int(time.time()*1000)%100000}”,
name=name, arguments=args)]terminar_reason=”tool_calls”, uso=uso, ) has_digit = bool(re.search(r”\d”, usuario)) want_math = has_digit y ( bool(re.search(r”[\+\-\*\/\^]”, usuario)) o “sqrt” en usuario o “raíz cuadrada” en usuario o cualquiera(w en usuario para w en [“calculate”, “compute”, “evaluate”, “what is”, “what’s”])) si “calculadora” en nombres_herramientas y quiere_matemáticas y no self._already_ Call (mensajes, “calculadora”): devuelve llamada (“calculadora”, {“expresión”: self._extract_math(usuario)}) si “get_current_time” en nombres_herramientas y no self._already_Call (mensajes, “get_current_time”): si hay alguno (w en usuario para w en [“time”, “date”, “today”, “now”, “o’clock”]): tz = “UTC” m = re.búsqueda(r”in ([a-zA-Z_\/ ]+)”, usuario) if m: cand = m.group(1).strip().title().replace(” “, “_”) tz = {“Tokyo”: “Asia/Tokyo”, “Delhi”: “Asia/Kolkata”, “New_York”: “America/New_York”, “Londres”: “Europa/Londres”}.get(cand, cand) return call(“get_current_time”, {“timezone”: tz}) si “recordar_hecho” en nombres_herramientas y no en self._already_ceived(messages, “remember_fact”): m = re.search(r”mi favorito (?:programación)? El idioma es (\w+)”, usuario) if m: return call(“remember_fact”, {“key”: “favorite_language”, “value”: m.group(1)}) if “recall_fact” en nombres_herramientas y no self._already_call(messages, “recall_fact”): si existe (w en usuario para w en [“my favorite”, “do you remember”, “recall”, “what did i tell”]): clave = “idioma_favorito” si “idioma” en el usuario else “nota” return call(“recall_fact”, {“key”: key}) si “run_python” en nombres_herramientas y no en self._already_call(messages, “run_python”): py_kw = any(w en usuario para w en [“fibonacci”, “prime”, “factorial”, “simulate”]) py_action = “python” en usuario y cualquiera (w en usuario para w en [“run”, “write”, “code”, “print”, “execute”, “snippet”]) si py_kw o py_action: si “fibonacci” en usuario: código = (“def fib(n):\na,b=0,1\n out=[]\n” ” para _ en rango(n):\n out.append(a); a,b=b,a+b\n return out\n” “print(fib(12))”) elif “primo” en usuario: código = (“primos=[n for n in range(2,50) ”
“if all(n%d for d in range(2,int(n**0.5)+1))]\nprint(primos)”) elif “factorial” en usuario: código = “importar matemáticas; print(math.factorial(10))” else: code = “print(sum(range(1,101)))” return call(“run_python”, {“code”: code}) if “web_search” en nombres_herramientas y no self._already_ceived(messages, “web_search”): si existe (w en usuario para w en [“search”, “look up”, “latest”, “news about”, “find information”]): return call(“web_search”, {“query”: self._last_user_text(messages)}) si corresponde (p en usuario para p en [“my name”, “who am i”, “what do i love”, “what i love”]): nombre, amor = self._scan_memory(mensajes) bits = []
if nombre: bits.append(f”tu nombre es {nombre}”) if amor: bits.append(f”te encanta {amor}”) if bits: return LLMResponse(content=”De nuestra conversación, ” + ” y “.join(bits) + “.”, tool_calls=[]terminar_reason=”parar”, uso=uso) salidas_herramientas = [m[“content”] for m en mensajes if m.get(“role”) == “tool”]if tool_outputs: unido = ” “.join(tool_outputs) respuesta = f”Según los resultados de la herramienta, esto es lo que encontré: {joined}” elif any(w en usuario para w en [“hello”, “hi”, “hey”]): respuesta = “¡Hola! Soy un agente nanobot simulado. Pídeme que calcule, diga la hora, ejecute Python o recuerde cosas”. más: respuesta = (“[mock LLM] Normalmente razonaría sobre esto con un modelo real. ” “Configure NANOBOT_API_KEY para usar un LLM en vivo. Por ahora, pruebe las indicaciones con matemáticas, ” “tiempo, Python o memoria para que pueda ver cómo se activa el bucle de la herramienta.”) return LLMResponse(content=answer, tool_calls=[]terminar_razón=”parar”, uso=uso)
llamadas_herramientas: lista[ToolCall] = campo(default_factory=lista) fin_reason: str = “detener” uso: Uso = campo(default_factory=Uso) clase Proveedor: “””Clase base. Un proveedor convierte (mensajes, herramientas) en una respuesta LLMR.””” nombre = “base” async def complete(self, mensajes: lista[dict]herramientas: lista[dict]) -> LLMResponse: elevar la clase NotImplementedError OpenAICompatibleProvider(Provider): “”” Funciona con OpenAI y todas las puertas de enlace compatibles con OpenAI (OpenRouter, DeepSeek, Together, vLLM, LM Studio, Ollama’s /v1, …). Esto refleja cómo nanobot habla con la mayoría de los proveedores bajo el capó. “”” name = “openai-compatible” def __init__(self, api_key: cadena, modelo: cadena, base_url: opcional[str] = Ninguno): de openai import AsyncOpenAI self.model = model self.client = AsyncOpenAI(api_key=api_key, base_url=base_url) async def complete(self, mensajes: lista[dict]herramientas: lista[dict]) -> LLMRespuesta: kwargs: dict[str, Any] = {“modelo”: self.model, “mensajes”: mensajes} si herramientas: kwargs[“tools”] = herramientas kwargs[“tool_choice”] = “auto” resp = await self.client.chat.completions.create(**kwargs) elección = resp.choices[0]
msg = elección.llamadas de mensaje: lista[ToolCall] = []
para tc en (msg.tool_calls o []): intente: args = json.loads(tc.function.arguments o “{}”) excepto json.JSONDecodeError: args = {“_raw”: tc.function.arguments} call.append(ToolCall(id=tc.id, name=tc.function.name, arguments=args)) uso = Uso( Prompt_tokens=getattr(resp.usage, “prompt_tokens”, 0) o 0, complete_tokens=getattr(resp.usage, “completion_tokens”, 0) o 0, ) return LLMResponse( content=msg.content, tool_calls=calls, Finish_reason=choice.finish_reason o “stop”, use=usage, ) clase MockProvider(Provider): “”” Un “LLM” determinista y basado en reglas, por lo que esto Todo el tutorial se ejecuta SIN clave API y SIN red, lo que le permite observar el bucle del agente, las llamadas a la herramienta y el funcionamiento de la memoria. Imita lo ÚNICO que importa para el bucle: decidir emitir una llamada a la herramienta (en la forma normalizada exacta que tendría un modelo real) y luego, una vez que se obtienen los resultados de la herramienta, producir una respuesta final en lenguaje natural. str = “mock-1”): self.model = modelo @staticmethod def _last_user_text(mensajes: lista[dict]) -> str: para m en reverso (mensajes): if m.get(“role”) == “usuario”: c = m.get(“content”) devuelve c si isinstance(c, str) else json.dumps(c) devuelve “” @staticmethod def _already_ceived(mensajes: lista[dict]nombre_herramienta: str) -> bool: para m en mensajes: si m.get(“role”) == “asistente” y m.get(“tool_calls”): para tc en m[“tool_calls”]: si tc[“function”][“name”] == nombre_herramienta: return True return False @staticmethod def _extract_math(text: str) -> str: “””Saque el primer fragmento matemático de una oración (ayudante solo simulado).””” t = re.sub(r”¿raíces cuadradas? of (\d+(?:\.\d+)?)”, r”sqrt(\1)”, text) t = t.replace(“^”, “**”) patrón = (r”(?:sqrt\(\d+(?:\.\d+)?\)|\d+(?:\.\d+)?)” r”(?:\s*(?:\*\*|[\+\-\*\/])\s*(?:sqrt\(\d+(?:\.\d+)?\)|\d+(?:\.\d+)?))*”) m = re.search(pattern, t) return m.group(0).strip() si m más t.strip() @staticmethod def _scan_memory(messages: list[dict]) -> tupla[Optional[str]Opcional[str]]: “””Leer datos simples de turnos de USUARIO anteriores: demuestra que la memoria de la sesión en realidad se está alimentando al modelo (conveniencia solo simulada).””” name = love = Ninguno para m en mensajes: if m.get(“role”) == “user” and isinstance(m.get(“content”), str): tx = m[“content”].lower() nm = re.search(r”mi nombre es (\w+)”, tx) if nm: nombre = nm.group(1).title() lv = re.search(r”i (?:love|like) (\w+)”, tx) if lv: love = lv.group(1).title() return nombre, love async def complete(self, mensajes: lista[dict]herramientas: lista[dict]) -> LLMResponse: await asyncio.sleep(0) usuario = self._last_user_text(messages).lower() nombres_herramientas = {t[“function”][“name”] para t en herramientas} uso = Uso(prompt_tokens=sum(len(str(m)) para m en mensajes) // 4, complete_tokens=12) def call(name, args): return LLMResponse( content=Ninguno, tool_calls=[ToolCall(id=f”call_{name}_{int(time.time()*1000)%100000}”,
name=name, arguments=args)]terminar_reason=”tool_calls”, uso=uso, ) has_digit = bool(re.search(r”\d”, usuario)) want_math = has_digit y ( bool(re.search(r”[\+\-\*\/\^]”, usuario)) o “sqrt” en usuario o “raíz cuadrada” en usuario o cualquiera(w en usuario para w en [“calculate”, “compute”, “evaluate”, “what is”, “what’s”])) si “calculadora” en nombres_herramientas y quiere_matemáticas y no self._already_ Call (mensajes, “calculadora”): devuelve llamada (“calculadora”, {“expresión”: self._extract_math(usuario)}) si “get_current_time” en nombres_herramientas y no self._already_Call (mensajes, “get_current_time”): si hay alguno (w en usuario para w en [“time”, “date”, “today”, “now”, “o’clock”]): tz = “UTC” m = re.búsqueda(r”in ([a-zA-Z_\/ ]+)”, usuario) if m: cand = m.group(1).strip().title().replace(” “, “_”) tz = {“Tokyo”: “Asia/Tokyo”, “Delhi”: “Asia/Kolkata”, “New_York”: “America/New_York”, “Londres”: “Europa/Londres”}.get(cand, cand) return call(“get_current_time”, {“timezone”: tz}) si “recordar_hecho” en nombres_herramientas y no en self._already_ceived(messages, “remember_fact”): m = re.search(r”mi favorito (?:programación)? El idioma es (\w+)”, usuario) if m: return call(“remember_fact”, {“key”: “favorite_language”, “value”: m.group(1)}) if “recall_fact” en nombres_herramientas y no self._already_call(messages, “recall_fact”): si existe (w en usuario para w en [“my favorite”, “do you remember”, “recall”, “what did i tell”]): clave = “idioma_favorito” si “idioma” en el usuario else “nota” return call(“recall_fact”, {“key”: key}) si “run_python” en nombres_herramientas y no en self._already_call(messages, “run_python”): py_kw = any(w en usuario para w en [“fibonacci”, “prime”, “factorial”, “simulate”]) py_action = “python” en usuario y cualquiera (w en usuario para w en [“run”, “write”, “code”, “print”, “execute”, “snippet”]) si py_kw o py_action: si “fibonacci” en usuario: código = (“def fib(n):\na,b=0,1\n out=[]\n” ” para _ en rango(n):\n out.append(a); a,b=b,a+b\n return out\n” “print(fib(12))”) elif “primo” en usuario: código = (“primos=[n for n in range(2,50) ”
“if all(n%d for d in range(2,int(n**0.5)+1))]\nprint(primos)”) elif “factorial” en usuario: código = “importar matemáticas; print(math.factorial(10))” else: code = “print(sum(range(1,101)))” return call(“run_python”, {“code”: code}) if “web_search” en nombres_herramientas y no self._already_ceived(messages, “web_search”): si existe (w en usuario para w en [“search”, “look up”, “latest”, “news about”, “find information”]): return call(“web_search”, {“query”: self._last_user_text(messages)}) si corresponde (p en usuario para p en [“my name”, “who am i”, “what do i love”, “what i love”]): nombre, amor = self._scan_memory(mensajes) bits = []
if nombre: bits.append(f”tu nombre es {nombre}”) if amor: bits.append(f”te encanta {amor}”) if bits: return LLMResponse(content=”De nuestra conversación, ” + ” y “.join(bits) + “.”, tool_calls=[]terminar_reason=”parar”, uso=uso) salidas_herramientas = [m[“content”] for m en mensajes if m.get(“role”) == “tool”]if tool_outputs: unido = ” “.join(tool_outputs) respuesta = f”Según los resultados de la herramienta, esto es lo que encontré: {joined}” elif any(w en usuario para w en [“hello”, “hi”, “hey”]): respuesta = “¡Hola! Soy un agente nanobot simulado. Pídeme que calcule, diga la hora, ejecute Python o recuerde cosas”. más: respuesta = (“[mock LLM] Normalmente razonaría sobre esto con un modelo real. ” “Configure NANOBOT_API_KEY para usar un LLM en vivo. Por ahora, pruebe las indicaciones con matemáticas, ” “tiempo, Python o memoria para que pueda ver cómo se activa el bucle de la herramienta.”) return LLMResponse(content=answer, tool_calls=[]terminar_razón=”parar”, uso=uso)