4
import subprocess, sys
def _pip_install(*pkgs):
strive:
subprocess.run([sys.executable, "-m", "pip", "install", "-q", *pkgs], verify=True)
besides Exception as e:
print(f"(pip set up skipped/failed for {pkgs}: {e})")
_HAVE_OPENAI = False
strive:
import openai
_HAVE_OPENAI = True
besides Exception:
_pip_install("openai>=1.0.0")
strive:
import openai
_HAVE_OPENAI = True
besides Exception:
_HAVE_OPENAI = False
strive:
import nest_asyncio
nest_asyncio.apply()
besides Exception:
strive:
_pip_install("nest_asyncio")
import nest_asyncio
nest_asyncio.apply()
besides Exception:
cross
import os
import re
import json
import time
import math
import asyncio
import examine
import textwrap
import contextlib
import io
from dataclasses import dataclass, subject
from typing import Any, Callable, Non-obligatory, Awaitable, get_type_hints
def banner(title: str) -> None:
line = "═" * 78
print(f"n{line}n {title}n{line}")
@dataclass
class ToolCall:
"""A normalized request from the mannequin to run one device."""
id: str
identify: str
arguments: dict
@dataclass
class Utilization:
prompt_tokens: int = 0
completion_tokens: int = 0
@property
def complete(self) -> int:
return self.prompt_tokens + self.completion_tokens
@dataclass
class LLMResponse:
"""The only form each supplier should return."""
content material: Non-obligatory[str]
tool_calls: record[ToolCall] = subject(default_factory=record)
finish_reason: str = "cease"
utilization: Utilization = subject(default_factory=Utilization)
class Supplier:
"""Base class. A supplier turns (messages, instruments) into an LLMResponse."""
identify = "base"
async def full(self, messages: record[dict], instruments: record[dict]) -> LLMResponse:
elevate NotImplementedError
class OpenAICompatibleProvider(Supplier):
"""
Works with OpenAI and each OpenAI-compatible gateway (OpenRouter, DeepSeek,
Collectively, vLLM, LM Studio, Ollama's /v1, ...). This mirrors how nanobot speaks
to most suppliers below the hood.
"""
identify = "openai-compatible"
def __init__(self, api_key: str, mannequin: str, base_url: Non-obligatory[str] = None):
from openai import AsyncOpenAI
self.mannequin = mannequin
self.consumer = AsyncOpenAI(api_key=api_key, base_url=base_url)
async def full(self, messages: record[dict], instruments: record[dict]) -> LLMResponse:
kwargs: dict[str, Any] = {"mannequin": self.mannequin, "messages": messages}
if instruments:
kwargs["tools"] = instruments
kwargs["tool_choice"] = "auto"
resp = await self.consumer.chat.completions.create(**kwargs)
selection = resp.selections[0]
msg = selection.message
calls: record[ToolCall] = []
for tc in (msg.tool_calls or []):
strive:
args = json.masses(tc.perform.arguments or "{}")
besides json.JSONDecodeError:
args = {"_raw": tc.perform.arguments}
calls.append(ToolCall(id=tc.id, identify=tc.perform.identify, arguments=args))
utilization = Utilization(
prompt_tokens=getattr(resp.utilization, "prompt_tokens", 0) or 0,
completion_tokens=getattr(resp.utilization, "completion_tokens", 0) or 0,
)
return LLMResponse(
content material=msg.content material,
tool_calls=calls,
finish_reason=selection.finish_reason or "cease",
utilization=utilization,
)
class MockProvider(Supplier):
"""
A deterministic, rule-based "LLM" so this whole tutorial runs with NO API key
and NO community — letting you watch the agent loop, device calls, and reminiscence work.
It imitates the ONE factor that issues for the loop: deciding to emit a device name
(within the precise normalized form an actual mannequin would) after which, as soon as device outcomes
come again, producing a closing natural-language reply. The agent loop can not inform
it aside from OpenAI — that is the entire level of the supplier contract.
"""
identify = "mock"
def __init__(self, mannequin: str = "mock-1"):
self.mannequin = mannequin
@staticmethod
def _last_user_text(messages: record[dict]) -> str:
for m in reversed(messages):
if m.get("function") == "person":
c = m.get("content material")
return c if isinstance(c, str) else json.dumps(c)
return ""
@staticmethod
def _already_called(messages: record[dict], tool_name: str) -> bool:
for m in messages:
if m.get("function") == "assistant" and m.get("tool_calls"):
for tc in m["tool_calls"]:
if tc["function"]["name"] == tool_name:
return True
return False
@staticmethod
def _extract_math(textual content: str) -> str:
"""Pull the primary math-looking chunk out of a sentence (mock-only helper)."""
t = re.sub(r"sq. roots? of (d+(?:.d+)?)", r"sqrt(1)", textual content)
t = t.substitute("^", "**")
sample = (r"(?:sqrt(d+(?:.d+)?)|d+(?:.d+)?)"
r"(?:s*(?:**|[+-*/])s*(?:sqrt(d+(?:.d+)?)|d+(?:.d+)?))*")
m = re.search(sample, t)
return m.group(0).strip() if m else t.strip()
@staticmethod
def _scan_memory(messages: record[dict]) -> tuple[Optional[str], Non-obligatory[str]]:
"""Learn again easy details from prior USER turns — proves session reminiscence is
truly being fed to the mannequin (mock-only comfort)."""
identify = love = None
for m in messages:
if m.get("function") == "person" and isinstance(m.get("content material"), str):
tx = m["content"].decrease()
nm = re.search(r"my identify is (w+)", tx)
if nm:
identify = nm.group(1).title()
lv = re.search(r"i (?:love|like) (w+)", tx)
if lv:
love = lv.group(1).title()
return identify, love
async def full(self, messages: record[dict], instruments: record[dict]) -> LLMResponse:
await asyncio.sleep(0)
person = self._last_user_text(messages).decrease()
tool_names = {t["function"]["name"] for t in instruments}
utilization = Utilization(prompt_tokens=sum(len(str(m)) for m in messages) // 4, completion_tokens=12)
def name(identify, args):
return LLMResponse(
content material=None,
tool_calls=[ToolCall(id=f"call_{name}_{int(time.time()*1000)%100000}",
name=name, arguments=args)],
finish_reason="tool_calls",
utilization=utilization,
)
has_digit = bool(re.search(r"d", person))
wants_math = has_digit and (
bool(re.search(r"[+-*/^]", person)) or "sqrt" in person
or "sq. root" in person
or any(w in person for w in ["calculate", "compute", "evaluate", "what is", "what's"]))
if "calculator" in tool_names and wants_math and never self._already_called(messages, "calculator"):
return name("calculator", {"expression": self._extract_math(person)})
if "get_current_time" in tool_names and never self._already_called(messages, "get_current_time"):
if any(w in person for w in ["time", "date", "today", "now", "o'clock"]):
tz = "UTC"
m = re.search(r"in ([a-zA-Z_/ ]+)", person)
if m:
cand = m.group(1).strip().title().substitute(" ", "_")
tz = {"Tokyo": "Asia/Tokyo", "Delhi": "Asia/Kolkata",
"New_York": "America/New_York", "London": "Europe/London"}.get(cand, cand)
return name("get_current_time", {"timezone": tz})
if "remember_fact" in tool_names and never self._already_called(messages, "remember_fact"):
m = re.search(r"my favourite (?:programming )?language is (w+)", person)
if m:
return name("remember_fact", {"key": "favorite_language", "worth": m.group(1)})
if "recall_fact" in tool_names and never self._already_called(messages, "recall_fact"):
if any(w in person for w in ["my favorite", "do you remember", "recall", "what did i tell"]):
key = "favorite_language" if "language" in person else "be aware"
return name("recall_fact", {"key": key})
if "run_python" in tool_names and never self._already_called(messages, "run_python"):
py_kw = any(w in person for w in ["fibonacci", "prime", "factorial", "simulate"])
py_action = "python" in person and any(
w in person for w in ["run", "write", "code", "print", "execute", "snippet"])
if py_kw or py_action:
if "fibonacci" in person:
code = ("def fib(n):n a,b=0,1n out=[]n"
" for _ in vary(n):n out.append(a); a,b=b,a+bn return outn"
"print(fib(12))")
elif "prime" in person:
code = ("primes=[n for n in range(2,50) "
"if all(n%d for d in range(2,int(n**0.5)+1))]nprint(primes)")
elif "factorial" in person:
code = "import math; print(math.factorial(10))"
else:
code = "print(sum(vary(1,101)))"
return name("run_python", {"code": code})
if "web_search" in tool_names and never self._already_called(messages, "web_search"):
if any(w in person for w in ["search", "look up", "latest", "news about", "find information"]):
return name("web_search", {"question": self._last_user_text(messages)})
if any(p in person for p in ["my name", "who am i", "what do i love", "what i love"]):
identify, love = self._scan_memory(messages)
bits = []
if identify:
bits.append(f"your identify is {identify}")
if love:
bits.append(f"you're keen on {love}")
if bits:
return LLMResponse(content material="From our dialog, " + " and ".be part of(bits) + ".",
tool_calls=[], finish_reason="cease", utilization=utilization)
tool_outputs = [m["content"] for m in messages if m.get("function") == "device"]
if tool_outputs:
joined = " ".be part of(tool_outputs)
reply = f"Primarily based on the device outcomes, this is what I discovered: {joined}"
elif any(w in person for w in ["hello", "hi", "hey"]):
reply = "Whats up! I am a mock nanobot agent. Ask me to calculate, inform time, run Python, or bear in mind issues."
else:
reply = ("[mock LLM] I might usually cause about this with an actual mannequin. "
"Set NANOBOT_API_KEY to make use of a dwell LLM. For now, strive prompts with math, "
"time, Python, or reminiscence so you may see the device loop hearth.")
return LLMResponse(content material=reply, tool_calls=[], finish_reason="cease", utilization=utilization)
