ツール呼び出し、セッションメモリ、スキル、MCP サーバーを備えたナノボット型 AI エージェントを Google Colab で構築するチュートリアル
このチュートリアルは、外部フレームワークに依存せず、Google Colab上でナノボット型 AI エージェントの核心コンポーネント（プロバイダー抽象化、ツール呼び出し、セッションメモリ、MCP サーバー）をゼロから実装する方法を詳述している。
キーポイント
フレームワーク依存からの脱却とコア理解
既存の外部エージェントフレームワークを使用せず、メッセージフロー、ツール呼び出し、メモリ管理などの基本ブロックを自ら再構築することで、AI エージェントの内部動作原理を明確に可視化する。
プロバイダー抽象化の実装
OpenAI や OpenRouter、Ollama など多様な LLM プロバイダーに対応可能な統一されたインターフェース（Provider Abstraction）を実装し、モデル間の互換性を確保する。
MCP スタイルツールサーバーとスキル
Model Context Protocol (MCP) に準拠したツールサーバーの構築や、エージェントに付与される「スキル」の定義方法など、拡張性の高い機能を実装する手順を示す。
Google Colab での完全実行環境
複雑なセットアップを避け、誰でもブラウザ上で即座にコードを実行・検証できる Google Colab 環境に最適化された軽量な実装例を提供する。
プロバイダー抽象化層の構築
OpenAI や Ollama など多様な LLM プロバイダーに共通するインターフェース（`Provider` クラス）を定義し、ツール呼び出しやトークン使用量を統一的なデータ構造（`LLMResponse`, `ToolCall`）で扱えるように設計しています。
API キー不要のモック環境
学習用として、実際の API 呼び出しを行わずにツール呼び出しのロジックやセッションメモリ機能をシミュレートする `MockProvider` を実装し、ネットワーク接続なしでエージェントループの動作を確認できるようにしています。
動的なツール選択と実行
ユーザーの入力内容（数式、時間、Python コードなど）を解析して自動的に適切なツール（計算機、時計、コード実行など）を選択し、その結果を会話履歴に追加することで、文脈に応じた回答を生成するループを実装しています。
影響分析・編集コメントを表示
影響分析
この記事は、AI エージェント開発における「ブラックボックス化」への対抗策として、基礎的なアーキテクチャを自ら実装する価値を再認識させる内容です。開発者がフレームワークの依存度を下げ、エージェントがどのように思考し、ツールを呼び出し、記憶を保持するかという本質的なメカニズムを深く理解することで、より堅牢でカスタマイズ可能な次世代エージェントの構築が可能になります。
編集コメント
既存のフレームワークに頼らず、AI エージェントの骨格を自ら組むことで深い理解を得られる貴重な実践例です。特に MCP 標準への対応やプロバイダー抽象化の手法は、実務での柔軟なシステム設計に直結する重要な知見を含んでいます。
このチュートリアルでは、ナノボットの核心アーキテクチャに触発された軽量なパーソナル AI エージェントを構築します。すべての部分が理解可能で Google Colab で実行可能であることを維持しながらです。まずプロバイダーの抽象化から始め、ツール登録、セッションメモリ、ライフサイクルフック、スキル、そして MCP スタイルのツールサーバーへと進みます。進行するにつれて、単に外部のエージェントフレームワークを使用するのではなく、コアとなる構築ブロックを自ら再作成します。これにより、メッセージ、ツール、メモリ、モデル応答が実用的なエージェントループ内でどのように連携するかを明確に理解できるようになります。
プロバイダーの抽象化とモック LLM の構築
コードをコピーしました（コピー済み）
異なるブラウザを使用してください
import subprocess, sys
def _pip_install(*pkgs):
try:
subprocess.run([sys.executable, "-m", "pip", "install", "-q", *pkgs], check=True)
except Exception as e:
print(f"(pip install skipped/failed for {pkgs}: {e})")
_HAVE_OPENAI = False
try:
import openai
_HAVE_OPENAI = True
except Exception:
_pip_install("openai>=1.0.0")
try:
import openai
_HAVE_OPENAI = True
except Exception:
_HAVE_OPENAI = False
try:
import nest_asyncio
nest_asyncio.apply()
except Exception:
try:
_pip_install("nest_asyncio")
import nest_asyncio
nest_asyncio.apply()
except Exception:
pass
import os
import re
import json
import time
import math
import asyncio
import inspect
import textwrap
import contextlib
import io
from dataclasses import dataclass, field
from typing import Any, Callable, Optional, Awaitable, get_type_hints
def banner(title: str) -> None:
line = "═" * 78
print(f"\n{line}\n {title}\n{line}")
@dataclass
class ToolCall:
"""モデルからツールを実行するための正規化されたリクエスト。"""
id: str
name: str
arguments: dict
@dataclass
class Usage:
prompt_tokens: int = 0
completion_tokens: int = 0
@property
def total(self) -> int:
return self.prompt_tokens + self.completion_tokens
@dataclass
class LLMResponse:
"""すべてのプロバイダーが返すべき単一の形状。"""
content: Optional[str]
tool_calls: list[ToolCall] = field(default_factory=list)
finish_reason: str = "stop"
usage: Usage = field(default_factory=Usage)
class Provider:
"""基底クラス。プロバイダーは (メッセージ, ツール) を LLMResponse に変換します。"""
name = "base"
async def complete(self, messages: list[dict], tools: list[dict]) -> LLMResponse:
raise NotImplementedError
class OpenAICompatibleProvider(Provider):
"""
OpenAI およびすべての OpenAI 互換ゲートウェイ (OpenRouter, DeepSeek,
Together, vLLM, LM Studio, Ollama の /v1 など) で動作します。これは
ナノボットが内部でほとんどのプロバイダーと通信する方法を模倣したものです。
"""
name = "openai-compatible"
def __init__(self, api_key: str, model: str, base_url: Optional[str] = None):
from openai import AsyncOpenAI
self.model = model
self.client = AsyncOpenAI(api_key=api_key, base_url=base_url)
async def complete(self, messages: list[dict], tools: list[dict]) -> LLMResponse:
kwargs: dict[str, Any] = {"model": self.model, "messages": messages}
if tools:
kwargs["tools"] = tools
kwargs["tool_choice"] = "auto"
resp = await self.client.chat.completions.create(**kwargs)
choice = resp.choices[0]
msg = choice.message
calls: list[ToolCall] = []
for tc in (msg.tool_calls or []):
try:
args = json.loads(tc.function.arguments or "{}")
except json.JSONDecodeError:
args = {"_raw": tc.function.arguments}
calls.append(ToolCall(id=tc.id, name=tc.function.name, arguments=args))
usage = Usage(
prompt_tokens=getattr(resp.usage, "prompt_tokens", 0) or 0,
completion_tokens=getattr(resp.usage, "completion_tokens", 0) or 0,
)
return LLMResponse(
content=msg.content,
tool_calls=calls,
finish_reason=choice.finish_reason or "stop",
usage=usage,
)
class MockProvider(Provider):
"""
このチュートリアルが API キーもネットワークも不要で実行できるようにする、
決定論的かつルールベースの「LLM」です。これにより、エージェントループ、
ツール呼び出し、メモリの動作を視聴できます。
ループにとって重要なのはたった一つのこと、つまりツール呼び出しを発行する
(実際のモデルが取る正規化された形状で) ことです。そして、ツール結果が
戻ってきたら、最終的な自然言語の回答を生成します。エージェントループは
これを OpenAI と区別できません — それがプロバイダー契約の全目的です。
"""
name = "mock"
def __init__(self, model: str = "mock-1"):
self.model = model
@staticmethod
def _last_user_text(messages: list[dict]) -> str:
for m in reversed(messages):
if m.get("role") == "user":
c = m.get("content")
return c if isinstance(c, str) else json.dumps(c)
return ""
@staticmethod
def _already_called(messages: list[dict], tool_name: str) -> bool:
for m in messages:
if m.get("role") == "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(text: str) -> str:
"""文から最初の数式のようなチャンクを抽出する (モック専用ヘルパー)。"""
t = re.sub(r"square roots? of (\d+(?:\.\d+)?)", r"sqrt(\1)", text)
t = t.replace("^", "**")
pattern = (r"(?:sqrt\(\d+(?:\.\d+)?\)|\d+(?:\.\d+)?)"
r"(?:\s*(?:\*\*|[\+\-\*\/])\s*(?:sqrt\(\d+(?:\.\d+)?\)|\d+(?:\.\d+)?))*")
m = re.search(pattern, t)
return m.group(0).strip() if m else t.strip()
@staticmethod
def _scan_memory(messages: list[dict]) -> tuple[Optional[str], Optional[str]]:
"""以前のユーザーのターンから単純な事実を読み取る — セッションメモリの
提供が実際にモデルにフィードされていることを証明する (モック専用便利機能)。"""
name = love = None
for m in messages:
if m.get("role") == "user" and isinstance(m.get("content"), str):
tx = m["content"].lower()
nm = re.search(r"my name is (\w+)", tx)
if nm:
name = nm.group(1).title()
lv = re.search(r"i (?:love|like) (\w+)", tx)
if lv:
love = lv.group(1).title()
return name, love
async def complete(self, messages: list[dict], tools: list[dict]) -> LLMResponse:
await asyncio.sleep(0)
user = self._last_user_text(messages).lower()
tool_names = {t["function"]["name"] for t in tools}
usage = Usage(prompt_tokens=sum(len(str(m)) for m in messages) // 4, completion_tokens=12)
def call(name, args):
return LLMResponse(
content=None,
tool_calls=[ToolCall(id=f"call_{name}_{int(time.time()*1000)%100000}",
name=name, arguments=args)],
finish_reason="tool_calls",
usage=usage,
)
has_digit = bool(re.search(r"\d", user))
wants_math = has_digit and (
bool(re.search(r"[\+\-\*\/\^]", user)) or "sqrt" in user
or "square root" in user
or any(w in user for w in ["calculate", "compute", "evaluate", "what is", "what's"]))
if "calculator" in tool_names and wants_math and not self._already_called(messages, "calculator"):
return call("calculator", {"expression": self._extract_math(user)})
if "get_current_time" in tool_names and not self._already_called(messages, "get_current_time"):
if any(w in user for w in ["time", "date", "today", "now", "o'clock"]):
tz = "UTC"
m = re.search(r"in ([a-zA-Z_\/ ]+)", user)
if m:
cand = m.group(1).strip().title().replace(" ", "_")
tz = {"Tokyo": "Asia/Tokyo", "Delhi": "Asia/Kolkata",
"New_York": "America/New_York", "London": "Europe/London"}.get(cand, cand)
return call("get_current_time", {"timezone": tz})
if "remember_fact" in tool_names and not self._already_called(messages, "remember_fact"):
m = re.search(r"my favorite (?:programming )?language is (\w+)", user)
if m:
return call("remember_fact", {"key": "favorite_language", "value": m.group(1)})
if "recall_fact" in tool_names and not self._already_called(messages, "recall_fact"):
if any(w in user for w in ["my favorite", "do you remember", "recall", "what did i tell"]):
key = "favorite_language" if "language" in user else "note"
return call("recall_fact", {"key": key})
if "run_python" in tool_names and not self._already_called(messages, "run_python"):
py_kw = any(w in user for w in ["fibonacci", "prime", "factorial", "simulate"])
py_action = "python" in user and any(
w in user for w in ["run", "write", "code", "print", "execute", "snippet"])
if py_kw or py_action:
if "fibonacci" in user:
code = ("def fib(n):\n a,b=0,1\n out=[]\n"
" for _ in range(n):\n out.append(a); a,b=b,a+b\n return out\n"
"print(fib(12))")
elif "prime" in user:
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 user:
code = "import math; print(math.factorial(10))"
else:
code = "print(sum(range(1,101)))"
return call("run_python", {"code": code})
if "web_search" in tool_names and not self._already_called(messages, "web_search"):
if any(w in user for w in ["search", "look up", "latest", "news about", "find information"]):
return call("web_search", {"query": self._last_user_text(messages)})
if any(p in user for p in ["my name", "who am i", "what do i love", "what i love"]):
name, love = self._scan_memory(messages)
bits = []
if name:
bits.append(f"your name is {name}")
if love:
bits.append(f"you love {love}")
if bits:
return LLMResponse(content="From our conversation, " + " and ".join(bits) + ".",
tool_calls=[], finish_reason="stop", usage=usage)
tool_outputs = [m["content"] for m in messages if m.get("role") == "tool"]
if tool_outputs:
joined = " ".join(tool_outputs)
answer = f"Based on the tool results, here's what I found: {joined}"
elif any(w in user for w in ["hello", "hi", "hey"]):
answer = "Hello! I'm a mock nanobot agent. Ask me to calculate, tell time, run Python, or remember things."
else:
answer = ("[mock LLM] I would normally reason about this with a real model. "
"Set NANOBOT_API_KEY to use a live LLM. For now, try prompts with math, "
"time, Python, or memory so you can see the tool loop fire.")
return LLMResponse(content=answer, tool_calls=[], finish_reason="stop", usage=usage)
環境のセットアップ、オプション依存関係のインストール、および本チュートリアルの全体に必要なインポートの準備を行います。ここでは、エージェントが実際の OpenAI 互換モデルまたは決定論的なモックプロバイダーのいずれとも連携できるようにするプロバイダ抽象化を定義します。また、残りのエージェントループがバックエンドモデルに依存せずに動作できるよう、正規化されたレスポンス構造も構築します。
ツールレジストリとトークン予算制約付きメモリの作成
コードをコピーしました（別のブラウザを使用）
_PYTYPE_TO_JSON = {str: "string", int: "integer", float: "number", bool: "boolean",
list: "array", dict: "object"}
@dataclass
class Tool:
name: str
description: str
parameters: dict
func: Callable
is_async: bool
def spec(self) -> dict:
"""OpenAI 風ツール仕様モデルが参照するもの。"""
return {"type": "function",
"function": {"name": self.name,
"description": self.description,
"parameters": self.parameters}}
async def __call__(self, **kwargs) -> str:
try:
result = self.func(**kwargs)
if inspect.isawaitable(result):
result = await result
return result if isinstance(result, str) else json.dumps(result, default=str)
except Exception as e:
return f"ERROR running tool '{self.name}': {type(e).__name__}: {e}"
def tool(func: Optional[Callable] = None, *, name: Optional[str] = None):
"""
通常の関数をツールに変換するデコレータ。JSON スキーマは型ヒントとドキュメントストリングの最初の行から導出されます。
パラメータの説明は、ドキュメントストリング内の単純な 'param: description' ブロックで追加できます。
例:
@tool
def calculator(expression: str) -> str:
'''数式を評価して結果を返します。
expression: "2 + 2 * 3" や "sqrt(16)" のような数式'''
...
"""
def make(f: Callable) -> Tool:
hints = get_type_hints(f)
sig = inspect.signature(f)
doc = inspect.getdoc(f) or ""
summary = doc.split("\n", 1)[0].strip() or f.__name__
param_docs: dict[str, str] = {}
for line in doc.splitlines()[1:]:
m = re.match(r"\s*(\w+)\s*:\s*(.+)", line)
if m and m.group(1) in sig.parameters:
param_docs[m.group(1)] = m.group(2).strip()
props, required = {}, []
for pname, p in sig.parameters.items():
if pname == "self":
continue
jtype = _PYTYPE_TO_JSON.get(hints.get(pname, str), "string")
schema = {"type": jtype}
if pname in param_docs:
schema["description"] = param_docs[pname]
props[pname] = schema
if p.default is inspect.Parameter.empty:
required.append(pname)
parameters = {"type": "object", "properties": props, "required": required}
return Tool(name=name or f.__name__, description=summary,
parameters=parameters, func=f, is_async=inspect.iscoroutinefunction(f))
return make(func) if func else make
class ToolRegistry:
def __init__(self):
self._tools: dict[str, Tool] = {}
def add(self, t: Tool) -> None:
self._tools[t.name] = t
def add_function(self, f: Callable) -> None:
self.add(tool(f))
def get(self, name: str) -> Optional[Tool]:
return self._tools.get(name)
def specs(self) -> list[dict]:
return [t.spec() for t in self._tools.values()]
def names(self) -> list[str]:
return list(self._tools)
@tool
def calculator(expression: str) -> str:
"""算術式を評価して数値結果を返します。
expression: 数式、例：'2 + 2 * 3', 'sqrt(16)', '2 ** 10'"""
allowed = {k: getattr(math, k) for k in dir(math) if not k.startswith("_")}
allowed.update({"abs": abs, "round": round, "min": min, "max": max, "sqrt": math.sqrt})
expr = expression.replace("^", "**")
value = eval(expr, {"__builtins__": {}}, allowed)
return f"{expression} = {value}"
@tool
def get_current_time(timezone: str = "UTC") -> str:
"""IANA 時刻区分名に対する現在の日時を返します。
timezone: IANA タイムゾーン、例：'UTC', 'Asia/Tokyo', 'Asia/Kolkata', 'America/New_York'"""
from datetime import datetime
try:
from zoneinfo import ZoneInfo
now = datetime.now(ZoneInfo(timezone))
except Exception:
from datetime import timezone as _tz
now = datetime.now(_tz.utc)
timezone = "UTC (fallback)"
return f"Current time in {timezone}: "
@tool
def run_python(code: str) -> str:
"""制限された名前空間で短い Python スニペットを実行し、その標準出力を返します。
code: 実行する Python ソースコード; 出力を得るには print(...) を使用してください"""
safe_builtins = {"print": print, "range": range, "len": len, "sum": sum, "min": min,
"max": max, "abs": abs, "sorted": sorted, "enumerate": enumerate,
"list": list, "dict": dict, "set": set, "str": str, "int": int,
"float": float, "bool": bool, "map": map, "filter": filter,
"zip": zip, "all": all, "any": any, "round": round}
import math as _m
g = {"__builtins__": safe_builtins, "math": _m}
buf = io.StringIO()
try:
with contextlib.redirect_stdout(buf):
exec(code, g, {})
out = buf.getvalue().strip()
return f"stdout:\n{out}" if out else "(ran successfully, no stdout)"
except Exception as e:
return f"Python error: {type(e).__name__}: {e}"
@tool
def web_search(query: str) -> str:
"""クエリでウェブを検索し、短い結果スニペットを返します（スタブ）。
query: 検索クエリ文字列"""
return (f"[stub results for '{query}'] (1) Overview article. (2) Official docs. "
f"(3) Recent discussion. Swap web_search's body for a real API in production.")
def estimate_tokens(messages: list[dict]) -> int:
"""大まかなトークン推定（約 4 文字/トークン）— デモの予算管理には十分です。"""
chars = 0
for m in messages:
chars += len(str(m.get("content") or ""))
for tc in (m.get("tool_calls") or []):
chars += len(json.dumps(tc))
return max(1, chars // 4)
class Memory:
def __init__(self, token_budget: int = 3000):
self.token_budget = token_budget
self._sessions: dict[str, list[dict]] = {}
def history(self, session_key: str) -> list[dict]:
return self._sessions.setdefault(session_key, [])
def append(self, session_key: str, message: dict) -> None:
self.history(session_key).append(message)
def extend(self, session_key: str, messages: list[dict]) -> None:
self.history(session_key).extend(messages)
def compact(self, session_key: str) -> int:
"""トークン予算を下回るまで最も古いメッセージを削除します。削除された数を返します。
ツール呼び出し/ツール結果のペアの一貫性を保つため、先頭から完全なターン単位でトリミングします。（nanobot は要約も行いますが、明確さのためここではトリミングのみを実装しています。）"""
hist = self.history(session_key)
dropped = 0
while estimate_tokens(hist) > self.token_budget and len(hist) > 2:
hist.pop(0)
dropped += 1
while hist and hist[0].get("role") == "tool":
hist.pop(0); dropped += 1
return dropped
通常の Python 関数を呼び出し可能なエージェントツールに変換できるツールシステムを作成します。型ヒントとドキュストリング（docstrings）を使用して JSON スタイルのツールスキーマを自動的に生成することで、フレームワークの拡張性を高めています。また、電卓、時刻照会ツール、Python 実行ツール、ウェブ検索スタブ、トークン制限付きメモリといった実用的なオフラインツールも追加しています。
ライフサイクルフック、スキル、およびエージェントループの実装
コードをコピーしました（別のブラウザを使用）
@dataclass
class AgentHookContext:
iteration: int = 0
messages: list[dict] = field(default_factory=list)
response: Optional[LLMResponse] = None
usage: Usage = field(default_factory=Usage)
tool_calls: list[ToolCall] = field(default_factory=list)
tool_results: list[str] = field(default_factory=list)
final_content: Optional[str] = None
stop_reason: Optional[str] = None
error: Optional[Exception] = None
class AgentHook:
"""必要な部分をサブクラス化してオーバーライドしてください。すべての非同期メソッドはベストエフォートであり、隔離されています（1 つのフックが失敗してもエージェントはクラッシュしません）。"""
def wants_streaming(self) -> bool:
return False
async def before_iteration(self, context: AgentHookContext) -> None: ...
async def on_stream(self, context: AgentHookContext, delta: str) -> None: ...
async def on_stream_end(self, context: AgentHookContext, *, resuming: bool) -> None: ...
async def before_execute_tools(self, context: Ag
⟦CODE_0⟧
原文を表示
In this tutorial, we build a lightweight personal AI agent inspired by the core architecture of nanobot, while keeping every part understandable and runnable in Google Colab. We start from the provider abstraction, then move through tool registration, session memory, lifecycle hooks, skills, and an MCP-style tool server. As we progress, we do not just use an external agent framework; we recreate the core building blocks ourselves so we can clearly see how messages, tools, memory, and model responses work together within a practical agent loop.
Building the Provider Abstraction and Mock LLM
Copy CodeCopiedUse a different Browser
import subprocess, sys
def _pip_install(*pkgs):
try:
subprocess.run([sys.executable, "-m", "pip", "install", "-q", *pkgs], check=True)
except Exception as e:
print(f"(pip install skipped/failed for {pkgs}: {e})")
_HAVE_OPENAI = False
try:
import openai
_HAVE_OPENAI = True
except Exception:
_pip_install("openai>=1.0.0")
try:
import openai
_HAVE_OPENAI = True
except Exception:
_HAVE_OPENAI = False
try:
import nest_asyncio
nest_asyncio.apply()
except Exception:
try:
_pip_install("nest_asyncio")
import nest_asyncio
nest_asyncio.apply()
except Exception:
pass
import os
import re
import json
import time
import math
import asyncio
import inspect
import textwrap
import contextlib
import io
from dataclasses import dataclass, field
from typing import Any, Callable, Optional, 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 model to run one tool."""
id: str
name: str
arguments: dict
@dataclass
class Usage:
prompt_tokens: int = 0
completion_tokens: int = 0
@property
def total(self) -> int:
return self.prompt_tokens + self.completion_tokens
@dataclass
class LLMResponse:
"""The single shape every provider must return."""
content: Optional[str]
tool_calls: list[ToolCall] = field(default_factory=list)
finish_reason: str = "stop"
usage: Usage = field(default_factory=Usage)
class Provider:
"""Base class. A provider turns (messages, tools) into an LLMResponse."""
name = "base"
async def complete(self, messages: list[dict], tools: list[dict]) -> LLMResponse:
raise NotImplementedError
class OpenAICompatibleProvider(Provider):
"""
Works with OpenAI and every OpenAI-compatible gateway (OpenRouter, DeepSeek,
Together, vLLM, LM Studio, Ollama's /v1, ...). This mirrors how nanobot speaks
to most providers under the hood.
"""
name = "openai-compatible"
def __init__(self, api_key: str, model: str, base_url: Optional[str] = None):
from openai import AsyncOpenAI
self.model = model
self.client = AsyncOpenAI(api_key=api_key, base_url=base_url)
async def complete(self, messages: list[dict], tools: list[dict]) -> LLMResponse:
kwargs: dict[str, Any] = {"model": self.model, "messages": messages}
if tools:
kwargs["tools"] = tools
kwargs["tool_choice"] = "auto"
resp = await self.client.chat.completions.create(**kwargs)
choice = resp.choices[0]
msg = choice.message
calls: list[ToolCall] = []
for tc in (msg.tool_calls or []):
try:
args = json.loads(tc.function.arguments or "{}")
except json.JSONDecodeError:
args = {"_raw": tc.function.arguments}
calls.append(ToolCall(id=tc.id, name=tc.function.name, arguments=args))
usage = Usage(
prompt_tokens=getattr(resp.usage, "prompt_tokens", 0) or 0,
completion_tokens=getattr(resp.usage, "completion_tokens", 0) or 0,
)
return LLMResponse(
content=msg.content,
tool_calls=calls,
finish_reason=choice.finish_reason or "stop",
usage=usage,
)
class MockProvider(Provider):
"""
A deterministic, rule-based "LLM" so this entire tutorial runs with NO API key
and NO network — letting you watch the agent loop, tool calls, and memory work.
It imitates the ONE thing that matters for the loop: deciding to emit a tool call
(in the exact normalized shape a real model would) and then, once tool results
come back, producing a final natural-language answer. The agent loop cannot tell
it apart from OpenAI — that's the whole point of the provider contract.
"""
name = "mock"
def __init__(self, model: str = "mock-1"):
self.model = model
@staticmethod
def _last_user_text(messages: list[dict]) -> str:
for m in reversed(messages):
if m.get("role") == "user":
c = m.get("content")
return c if isinstance(c, str) else json.dumps(c)
return ""
@staticmethod
def _already_called(messages: list[dict], tool_name: str) -> bool:
for m in messages:
if m.get("role") == "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(text: str) -> str:
"""Pull the first math-looking chunk out of a sentence (mock-only helper)."""
t = re.sub(r"square roots? of (\d+(?:\.\d+)?)", r"sqrt(\1)", text)
t = t.replace("^", "**")
pattern = (r"(?:sqrt\(\d+(?:\.\d+)?\)|\d+(?:\.\d+)?)"
r"(?:\s*(?:\*\*|[\+\-\*\/])\s*(?:sqrt\(\d+(?:\.\d+)?\)|\d+(?:\.\d+)?))*")
m = re.search(pattern, t)
return m.group(0).strip() if m else t.strip()
@staticmethod
def _scan_memory(messages: list[dict]) -> tuple[Optional[str], Optional[str]]:
"""Read back simple facts from prior USER turns — proves session memory is
actually being fed to the model (mock-only convenience)."""
name = love = None
for m in messages:
if m.get("role") == "user" and isinstance(m.get("content"), str):
tx = m["content"].lower()
nm = re.search(r"my name is (\w+)", tx)
if nm:
name = nm.group(1).title()
lv = re.search(r"i (?:love|like) (\w+)", tx)
if lv:
love = lv.group(1).title()
return name, love
async def complete(self, messages: list[dict], tools: list[dict]) -> LLMResponse:
await asyncio.sleep(0)
user = self._last_user_text(messages).lower()
tool_names = {t["function"]["name"] for t in tools}
usage = Usage(prompt_tokens=sum(len(str(m)) for m in messages) // 4, completion_tokens=12)
def call(name, args):
return LLMResponse(
content=None,
tool_calls=[ToolCall(id=f"call_{name}_{int(time.time()*1000)%100000}",
name=name, arguments=args)],
finish_reason="tool_calls",
usage=usage,
)
has_digit = bool(re.search(r"\d", user))
wants_math = has_digit and (
bool(re.search(r"[\+\-\*\/\^]", user)) or "sqrt" in user
or "square root" in user
or any(w in user for w in ["calculate", "compute", "evaluate", "what is", "what's"]))
if "calculator" in tool_names and wants_math and not self._already_called(messages, "calculator"):
return call("calculator", {"expression": self._extract_math(user)})
if "get_current_time" in tool_names and not self._already_called(messages, "get_current_time"):
if any(w in user for w in ["time", "date", "today", "now", "o'clock"]):
tz = "UTC"
m = re.search(r"in ([a-zA-Z_\/ ]+)", user)
if m:
cand = m.group(1).strip().title().replace(" ", "_")
tz = {"Tokyo": "Asia/Tokyo", "Delhi": "Asia/Kolkata",
"New_York": "America/New_York", "London": "Europe/London"}.get(cand, cand)
return call("get_current_time", {"timezone": tz})
if "remember_fact" in tool_names and not self._already_called(messages, "remember_fact"):
m = re.search(r"my favorite (?:programming )?language is (\w+)", user)
if m:
return call("remember_fact", {"key": "favorite_language", "value": m.group(1)})
if "recall_fact" in tool_names and not self._already_called(messages, "recall_fact"):
if any(w in user for w in ["my favorite", "do you remember", "recall", "what did i tell"]):
key = "favorite_language" if "language" in user else "note"
return call("recall_fact", {"key": key})
if "run_python" in tool_names and not self._already_called(messages, "run_python"):
py_kw = any(w in user for w in ["fibonacci", "prime", "factorial", "simulate"])
py_action = "python" in user and any(
w in user for w in ["run", "write", "code", "print", "execute", "snippet"])
if py_kw or py_action:
if "fibonacci" in user:
code = ("def fib(n):\n a,b=0,1\n out=[]\n"
" for _ in range(n):\n out.append(a); a,b=b,a+b\n return out\n"
"print(fib(12))")
elif "prime" in user:
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 user:
code = "import math; print(math.factorial(10))"
else:
code = "print(sum(range(1,101)))"
return call("run_python", {"code": code})
if "web_search" in tool_names and not self._already_called(messages, "web_search"):
if any(w in user for w in ["search", "look up", "latest", "news about", "find information"]):
return call("web_search", {"query": self._last_user_text(messages)})
if any(p in user for p in ["my name", "who am i", "what do i love", "what i love"]):
name, love = self._scan_memory(messages)
bits = []
if name:
bits.append(f"your name is {name}")
if love:
bits.append(f"you love {love}")
if bits:
return LLMResponse(content="From our conversation, " + " and ".join(bits) + ".",
tool_calls=[], finish_reason="stop", usage=usage)
tool_outputs = [m["content"] for m in messages if m.get("role") == "tool"]
if tool_outputs:
joined = " ".join(tool_outputs)
answer = f"Based on the tool results, here's what I found: {joined}"
elif any(w in user for w in ["hello", "hi", "hey"]):
answer = "Hello! I'm a mock nanobot agent. Ask me to calculate, tell time, run Python, or remember things."
else:
answer = ("[mock LLM] I would normally reason about this with a real model. "
"Set NANOBOT_API_KEY to use a live LLM. For now, try prompts with math, "
"time, Python, or memory so you can see the tool loop fire.")
return LLMResponse(content=answer, tool_calls=[], finish_reason="stop", usage=usage)
We set up the environment, install optional dependencies, and prepare the imports needed for the full tutorial. We define a provider abstraction that allows the agent to work with either a real OpenAI-compatible model or a deterministic mock provider. We also build the normalized response structures so the rest of the agent loop can work independently of the backend model.
Creating the Tool Registry and Token-Budgeted Memory
Copy CodeCopiedUse a different Browser
_PYTYPE_TO_JSON = {str: "string", int: "integer", float: "number", bool: "boolean",
list: "array", dict: "object"}
@dataclass
class Tool:
name: str
description: str
parameters: dict
func: Callable
is_async: bool
def spec(self) -> dict:
"""OpenAI-style tool spec the model sees."""
return {"type": "function",
"function": {"name": self.name,
"description": self.description,
"parameters": self.parameters}}
async def __call__(self, **kwargs) -> str:
try:
result = self.func(**kwargs)
if inspect.isawaitable(result):
result = await result
return result if isinstance(result, str) else json.dumps(result, default=str)
except Exception as e:
return f"ERROR running tool '{self.name}': {type(e).__name__}: {e}"
def tool(func: Optional[Callable] = None, *, name: Optional[str] = None):
"""
Decorator that turns a plain function into a Tool, deriving the JSON schema from
type hints and the first line of the docstring. Param descriptions can be added
with a simple 'param: description' block in the docstring.
Example:
@tool
def calculator(expression: str) -> str:
'''Evaluate a math expression and return the result.
expression: a math expression like "2 + 2 * 3" or "sqrt(16)"'''
...
"""
def make(f: Callable) -> Tool:
hints = get_type_hints(f)
sig = inspect.signature(f)
doc = inspect.getdoc(f) or ""
summary = doc.split("\n", 1)[0].strip() or f.__name__
param_docs: dict[str, str] = {}
for line in doc.splitlines()[1:]:
m = re.match(r"\s*(\w+)\s*:\s*(.+)", line)
if m and m.group(1) in sig.parameters:
param_docs[m.group(1)] = m.group(2).strip()
props, required = {}, []
for pname, p in sig.parameters.items():
if pname == "self":
continue
jtype = _PYTYPE_TO_JSON.get(hints.get(pname, str), "string")
schema = {"type": jtype}
if pname in param_docs:
schema["description"] = param_docs[pname]
props[pname] = schema
if p.default is inspect.Parameter.empty:
required.append(pname)
parameters = {"type": "object", "properties": props, "required": required}
return Tool(name=name or f.__name__, description=summary,
parameters=parameters, func=f, is_async=inspect.iscoroutinefunction(f))
return make(func) if func else make
class ToolRegistry:
def __init__(self):
self._tools: dict[str, Tool] = {}
def add(self, t: Tool) -> None:
self._tools[t.name] = t
def add_function(self, f: Callable) -> None:
self.add(tool(f))
def get(self, name: str) -> Optional[Tool]:
return self._tools.get(name)
def specs(self) -> list[dict]:
return [t.spec() for t in self._tools.values()]
def names(self) -> list[str]:
return list(self._tools)
@tool
def calculator(expression: str) -> str:
"""Evaluate an arithmetic expression and return the numeric result.
expression: a math expression, e.g. '2 + 2 * 3', 'sqrt(16)', '2 ** 10'"""
allowed = {k: getattr(math, k) for k in dir(math) if not k.startswith("_")}
allowed.update({"abs": abs, "round": round, "min": min, "max": max, "sqrt": math.sqrt})
expr = expression.replace("^", "**")
value = eval(expr, {"__builtins__": {}}, allowed)
return f"{expression} = {value}"
@tool
def get_current_time(timezone: str = "UTC") -> str:
"""Return the current date and time for an IANA timezone name.
timezone: IANA tz like 'UTC', 'Asia/Tokyo', 'Asia/Kolkata', 'America/New_York'"""
from datetime import datetime
try:
from zoneinfo import ZoneInfo
now = datetime.now(ZoneInfo(timezone))
except Exception:
from datetime import timezone as _tz
now = datetime.now(_tz.utc)
timezone = "UTC (fallback)"
return f"Current time in {timezone}: "
@tool
def run_python(code: str) -> str:
"""Execute a short Python snippet in a restricted namespace and return its stdout.
code: Python source code to run; use print(...) to produce output"""
safe_builtins = {"print": print, "range": range, "len": len, "sum": sum, "min": min,
"max": max, "abs": abs, "sorted": sorted, "enumerate": enumerate,
"list": list, "dict": dict, "set": set, "str": str, "int": int,
"float": float, "bool": bool, "map": map, "filter": filter,
"zip": zip, "all": all, "any": any, "round": round}
import math as _m
g = {"__builtins__": safe_builtins, "math": _m}
buf = io.StringIO()
try:
with contextlib.redirect_stdout(buf):
exec(code, g, {})
out = buf.getvalue().strip()
return f"stdout:\n{out}" if out else "(ran successfully, no stdout)"
except Exception as e:
return f"Python error: {type(e).__name__}: {e}"
@tool
def web_search(query: str) -> str:
"""Search the web for a query and return short result snippets (STUB).
query: the search query string"""
return (f"[stub results for '{query}'] (1) Overview article. (2) Official docs. "
f"(3) Recent discussion. Swap web_search's body for a real API in production.")
def estimate_tokens(messages: list[dict]) -> int:
"""Rough token estimate (~4 chars/token) — good enough for budgeting demos."""
chars = 0
for m in messages:
chars += len(str(m.get("content") or ""))
for tc in (m.get("tool_calls") or []):
chars += len(json.dumps(tc))
return max(1, chars // 4)
class Memory:
def __init__(self, token_budget: int = 3000):
self.token_budget = token_budget
self._sessions: dict[str, list[dict]] = {}
def history(self, session_key: str) -> list[dict]:
return self._sessions.setdefault(session_key, [])
def append(self, session_key: str, message: dict) -> None:
self.history(session_key).append(message)
def extend(self, session_key: str, messages: list[dict]) -> None:
self.history(session_key).extend(messages)
def compact(self, session_key: str) -> int:
"""Drop oldest messages until under the token budget. Returns #dropped.
Keeps tool-call/tool-result pairs consistent by trimming from the front in
whole turns. (nanobot also summarizes; we keep it to trimming for clarity.)"""
hist = self.history(session_key)
dropped = 0
while estimate_tokens(hist) > self.token_budget and len(hist) > 2:
hist.pop(0)
dropped += 1
while hist and hist[0].get("role") == "tool":
hist.pop(0); dropped += 1
return dropped
We create a tool system that allows ordinary Python functions to become callable agent tools. We use type hints and docstrings to automatically generate JSON-style tool schemas, which makes the framework easier to extend. We also add practical offline tools such as a calculator, a time lookup tool, a Python execution tool, a web search stub, and token-budgeted memory.
Implementing Lifecycle Hooks, Skills, and the Agent Loop
Copy CodeCopiedUse a different Browser
@dataclass
class AgentHookContext:
iteration: int = 0
messages: list[dict] = field(default_factory=list)
response: Optional[LLMResponse] = None
usage: Usage = field(default_factory=Usage)
tool_calls: list[ToolCall] = field(default_factory=list)
tool_results: list[str] = field(default_factory=list)
final_content: Optional[str] = None
stop_reason: Optional[str] = None
error: Optional[Exception] = None
class AgentHook:
"""Subclass and override what you need. All async methods are best-effort and
isolated (one failing hook won't crash the agent)."""
def wants_streaming(self) -> bool:
return False
async def before_iteration(self, context: AgentHookContext) -> None: ...
async def on_stream(self, context: AgentHookContext, delta: str) -> None: ...
async def on_stream_end(self, context: AgentHookContext, *, resuming: bool) -> None: ...
async def before_execute_tools(self, context: Ag
関連記事
今日のまとめ
AI日報で今日の重要ニュースをまとめ読み