上一篇讲了消息总线,两个 asyncio.Queue 把"消息入口"和"处理逻辑"解耦。看完应该能理解:新增一个聊天渠道(比如微信),不用改任何业务代码。
不过 Agent 真正干活还得靠另一个能力——调工具(Tool)。
没有 Tool 的 LLM 是个聊天机器。有 Tool 的 LLM 才能读文件、跑 shell、查数据库、发请求。这篇就讲 nanobot 怎么把"千差万别的能力"统一成同一个接口,以及怎么在不修改 nanobot 源码的前提下无限扩展。
对应教程仓库的 s02(Tool 抽象) + s03(Provider 抽象) 两个章节。
1. 第一个问题:Tool 怎么设计才不会被 if-else 淹没
假设 LLM 决定要读我电脑上一个文件。直觉的写法:
python
def handle_tool_call(name, args):
if name == "read_file":
return open(args["path"]).read()
elif name == "write_file":
with open(args["path"], "w") as f:
f.write(args["content"])
elif name == "run_shell":
return subprocess.run(args["cmd"], shell=True).stdout
elif name == "search_web":
return requests.get(args["url"]).text
# ... 加一个 tool 就要加一个 elif
加到第 5 个还能接受,加到第 10 个这个函数已经没人想动了。
更糟的是,每个新 Tool 还要让 LLM 知道它存在。你得把 Tool 的描述、参数 schema 喂给 LLM,否则 LLM 不知道有 read_file 这回事。每次加 Tool,你得改两处:handle_tool_call 函数 + 给 LLM 的 schema 列表。
nanobot 的解法:把"Tool 的描述"和"Tool 的实现"绑在一起,用一个注册表统一管。
2. nanobot的设计
python
class Tool:
"""所有 Tool 的基类。子类只需实现 4 个字段。"""
name: str = ""
description: str = ""
parameters: dict = {}
async def run(self, **kwargs) -> str:
raise NotImplementedError
class ReadFileTool(Tool):
name = "read_file"
description = "读取文件内容,返回 utf-8 文本"
parameters = {
"type": "object",
"properties": {
"path": {"type": "string", "description": "文件绝对或相对路径"},
},
"required": ["path"],
}
async def run(self, path: str) -> str:
try:
with open(path) as f:
return f.read()
except Exception as e:
return f"ERROR: {e}"
class ToolRegistry:
"""name -> Tool 的注册表。"""
def __init__(self):
self._tools: dict[str, Tool] = {}
def register(self, tool: Tool):
self._tools[tool.name] = tool
def to_anthropic_tools(self) -> list[dict]:
"""导出给 LLM 的 schema 列表。"""
return [
{"name": t.name, "description": t.description, "input_schema": t.parameters}
for t in self._tools.values()
]
async def dispatch(self, name: str, args: dict) -> str:
"""LLM 决定调哪个 tool 时,找到对应实例执行。"""
return await self._tools[name].run(**args)
这段代码解决了三个问题:
问题 1:怎么加新 Tool 不改 if-else?
答:写一个继承 Tool 的类,调用 registry.register(MyTool()) 就完事。dispatch 自动按 name 路由,不用动 registry 代码。
问题 2:怎么让 LLM 知道有哪些 Tool 可用?
答:调 registry.to_anthropic_tools(),自动把每个 Tool 的 name / description / parameters 拼成 LLM 能看懂的 schema。加新 Tool 不需要再维护一份独立的 schema 列表。
问题 3:怎么让"第三方写的新 Tool"被 nanobot 自动发现?
答:教程里提了一句 pkgutil + entry-point,s02 重点是设计思想。要看真实的 entry-point 注册可以翻 nanobot 真实代码的 nanobot/skills/ 目录。
3. 第二个问题:换 LLM 厂商要不要改业务代码
光有 Tool 还不够,Tool 是给 LLM 用的,但 LLM 本身要换怎么办?
Anthropic、OpenAI、DeepSeek、Gemini、智谱、阿里 DashScope、Moonshot、Groq、vLLM、OpenRouter……每家 API 都不一样。每家 SDK 的调用方式、返回结构、错误码都不同。
如果业务代码里到处都是 from openai import OpenAI,那想换到 DeepSeek 就得满世界改。
nanobot 的解法:所有 LLM 都接成同一个 LLMProvider 接口,业务代码只认这个接口。
s03 章节的简化版:
python
class LLMProvider(ABC):
"""所有 LLM 厂商适配层的基类。"""
@abstractmethod
async def chat(self, user_text: str) -> str:
raise NotImplementedError
class AnthropicProvider(LLMProvider):
def __init__(self, client: Anthropic | None = None):
self._client = client or _shared_client()
async def chat(self, user_text: str) -> str:
resp = self._client.messages.create(
model=MODEL_ID,
max_tokens=1024,
messages=[{"role": "user", "content": user_text}],
)
for b in resp.content:
if hasattr(b, "text"):
return b.text
return ""
业务代码拿到的是 LLMProvider,不知道背后是 Anthropic 直连还是走代理。
4. ProviderSpec:所有厂商的"身份证"
光有 LLMProvider 还不够——怎么知道支持哪些厂商?哪家要什么 env 变量?哪家要 base_url?
s03 的解法是维护一份元数据单一来源表:
python
@dataclass
class ProviderSpec:
name: str # spec id
display_name: str # 给人看的名字
env_key: str # 该 provider 需要的 env 变量
base_url: str | None # None = 走 SDK 默认;有值 = 走代理
PROVIDERS: list[ProviderSpec] = [
ProviderSpec(
name="anthropic",
display_name="Anthropic 直连",
env_key="ANTHROPIC_API_KEY",
base_url=None,
),
ProviderSpec(
name="proxy",
display_name="Anthropic 兼容代理(走 .env)",
env_key="ANTHROPIC_BASE_URL",
base_url=os.getenv("ANTHROPIC_BASE_URL"),
),
]
这份表是新增厂商的唯一入口。要加一个 DeepSeek?往 PROVIDERS 列表加一行;要加一个本地 Ollama?加一行 base_url 是 http://localhost:11434/v1 的 spec。
业务代码永远不写 if spec.name == “anthropic”。它只调 factory.create("deepseek"),拿到一个 LLMProvider 实例就能用。
5. 两个常见误区
5.1 误区 1:业务代码 import 具体 SDK
python
# 反例
from openai import OpenAI
client = OpenAI()
resp = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": text}],
)
这种写法有一个隐藏代价——想换 LLM 厂商时到处搜替换字符串。
正确做法:用 LLMProvider 抽象 + ProviderSpec 单一来源表:
业务代码永远不 import 具体 SDK,只调 factory.create("xxx").chat(...)切换 = 改环境变量( PROVIDER=deepseek),业务代码不动新增厂商 = 加一行 ProviderSpec,核心 Provider 实现一行不动
5.2 误区 2:Tool 不走注册表
python
# 反例
def handle_tool_call(name, args):
if name == "read_file": return open(args["path"]).read()
elif name == "write_file": ...
函数一多根本无法维护。
正确做法:继承 Tool 基类 + registry.register():
python
class MyTool(Tool):
name = "my_tool"
description = "..."
parameters = {...}
async def run(self, **kwargs):
...
registry.register(MyTool())
加新 Tool 真的就是"写一个类 + register 一下"两步。registry.to_anthropic_tools() 自动把 schema 喂给 LLM,registry.dispatch() 自动按 name 路由。
6. 下一篇
下一篇看 s04 + s05 章节,讲 Agent 状态机——一条用户消息是怎么走完 7 个状态、跑完一整轮 LLM 对话的。状态机是 nanobot 最核心的编排逻辑,把消息总线、Tool、Provider 三个机制串成完整的一轮对话。
7. 参考资料
nanobot 源码:https://github.com/HKUDS/nanobot nanobot-tutorial(14 章配套教程):https://github.com/yaoweizhang/nanobot_tutorial nanobot 官方 Roadmap:https://github.com/HKUDS/nanobot/discussions/431
跟读建议:跑 python s02_tool/code.py 体验一下"LLM 调 read_file 读文件"的完整流程,再对照 nanobot/skills/ 看真实 Tool 注册代码。
夜雨聆风