快速参考手册
nanobot 快速参考手册
🎯 适合已经了解基本概念,需要快速查找 API 和用法的开发者
目录
核心组件 API
AgentLoop
from nanobot.agent.loop import AgentLoop
from nanobot.bus.queue import MessageBus
from nanobot.providers.litellm_provider import LiteLLMProvider
from pathlib import Path
# 初始化
bus = MessageBus()
provider = LiteLLMProvider(api_key="sk-...", default_model="...")
agent = AgentLoop(
bus=bus,
provider=provider,
workspace=Path.home() / ".nanobot",
model="anthropic/claude-opus-4-5", # 可选,覆盖默认模型
max_iterations=20, # 可选,最大循环次数
brave_api_key="BSA-..." # 可选,Web 搜索
)
# 运行 Agent 循环(异步)
await agent.run()
# 直接处理消息(CLI 使用)
response = await agent.process_direct("你好", session_key="cli:test")
print(response)
# 停止循环
agent.stop()
ToolRegistry
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.agent.tools.base import Tool
# 创建注册表
registry = ToolRegistry()
# 注册工具
registry.register(MyTool())
# 获取工具
tool = registry.get("tool_name")
# 获取所有工具定义(用于 LLM)
definitions = registry.get_definitions()
# 执行工具
result = await registry.execute("tool_name", {"arg": "value"})
ContextBuilder
from nanobot.agent.context import ContextBuilder
from pathlib import Path
# 初始化
builder = ContextBuilder(workspace=Path.home() / ".nanobot")
# 构建系统提示
system_prompt = builder.build_system_prompt(skill_names=["github", "weather"])
# 构建完整消息列表
messages = builder.build_messages(
history=[
{"role": "user", "content": "之前的消息"},
{"role": "assistant", "content": "之前的回复"}
],
current_message="当前消息",
skill_names=["github"],
media=["/path/to/image.jpg"] # 可选,图片附件
)
# 添加工具结果
messages = builder.add_tool_result(
messages=messages,
tool_call_id="call_123",
tool_name="read_file",
result="文件内容..."
)
# 添加助手消息
messages = builder.add_assistant_message(
messages=messages,
content="回复内容",
tool_calls=[...] # 可选
)
SessionManager
from nanobot.session.manager import SessionManager
from pathlib import Path
# 初始化
manager = SessionManager(workspace=Path.home() / ".nanobot")
# 获取或创建会话
session = manager.get_or_create("telegram:123456")
# 添加消息
session.add_message("user", "用户消息")
session.add_message("assistant", "助手回复")
# 获取历史(LLM 格式)
history = session.get_history()
# 保存会话
manager.save(session)
MessageBus
from nanobot.bus.queue import MessageBus
from nanobot.bus.events import InboundMessage, OutboundMessage
# 创建总线
bus = MessageBus()
# 发布入站消息
await bus.publish_inbound(InboundMessage(
channel="cli",
sender_id="user",
chat_id="123",
content="消息内容",
session_key="cli:123",
media=["/path/to/image.jpg"] # 可选
))
# 消费入站消息(Agent 使用)
msg = await bus.consume_inbound()
# 发布出站消息
await bus.publish_outbound(OutboundMessage(
channel="telegram",
chat_id="123",
content="回复内容"
))
# 订阅出站消息(Channel 使用)
bus.subscribe_outbound("telegram", my_callback)
# 运行消息分发(后台任务)
await bus.dispatch_outbound()
CLI 命令速查
初始化
# 初始化配置和工作区
nanobot onboard
Agent 命令
# 单次对话
nanobot agent -m "你好,请介绍一下自己"
# 交互式对话
nanobot agent
# 指定会话 ID
nanobot agent -s my-session -m "继续之前的话题"
网关服务
# 启动网关(支持 Telegram/WhatsApp)
nanobot gateway
# 指定端口
nanobot gateway --port 8080
# 详细输出
nanobot gateway --verbose
渠道管理
# 查看渠道状态
nanobot channels status
# 登录 WhatsApp(扫描二维码)
nanobot channels login
定时任务
# 列出所有任务
nanobot cron list
# 添加任务(使用 cron 表达式)
nanobot cron add --name "morning" --message "早上好!" --cron "0 9 * * *"
# 添加任务(使用间隔秒数)
nanobot cron add --name "hourly" --message "检查状态" --every 3600
# 删除任务
nanobot cron remove <job_id>
# 启用/禁用任务
nanobot cron enable <job_id>
nanobot cron enable <job_id> --disable
# 手动运行任务
nanobot cron run <job_id>
状态查看
# 查看系统状态
nanobot status
配置文件模板
基础配置 (~/.nanobot/config.json)
{
"providers": {
"openrouter": {
"apiKey": "sk-or-v1-YOUR_KEY_HERE"
}
},
"agents": {
"defaults": {
"model": "anthropic/claude-opus-4-5"
}
},
"tools": {
"web": {
"search": {
"apiKey": "BSA-YOUR_BRAVE_API_KEY"
}
}
}
}
完整配置示例
{
"providers": {
"openrouter": {
"apiKey": "sk-or-v1-xxx"
},
"groq": {
"apiKey": "gsk_xxx"
},
"anthropic": {
"apiKey": "sk-ant-xxx"
},
"openai": {
"apiKey": "sk-xxx"
}
},
"agents": {
"defaults": {
"model": "anthropic/claude-opus-4-5",
"maxIterations": 20
}
},
"channels": {
"telegram": {
"enabled": true,
"token": "123456:ABC-DEF...",
"allowFrom": ["123456789"]
},
"whatsapp": {
"enabled": true,
"allowFrom": ["+1234567890"]
}
},
"tools": {
"web": {
"search": {
"apiKey": "BSA-xxx"
}
}
}
}
本地模型配置(vLLM)
{
"providers": {
"vllm": {
"apiKey": "dummy",
"apiBase": "http://localhost:8000/v1"
}
},
"agents": {
"defaults": {
"model": "meta-llama/Llama-3.1-8B-Instruct"
}
}
}
工具开发模板
基础工具模板
from typing import Any
from nanobot.agent.tools.base import Tool
class MyTool(Tool):
"""工具描述"""
@property
def name(self) -> str:
"""工具名称(LLM 调用时使用)"""
return "my_tool"
@property
def description(self) -> str:
"""工具功能描述(告诉 LLM 这个工具是做什么的)"""
return "这个工具可以做某某事情"
@property
def parameters(self) -> dict[str, Any]:
"""参数定义(JSON Schema 格式)"""
return {
"type": "object",
"properties": {
"arg1": {
"type": "string",
"description": "参数1的含义"
},
"arg2": {
"type": "integer",
"description": "参数2的含义",
"default": 10
}
},
"required": ["arg1"]
}
async def execute(self, arg1: str, arg2: int = 10) -> str:
"""
执行工具逻辑
参数:
arg1: 第一个参数
arg2: 第二个参数(可选)
返回:
执行结果(字符串)
"""
try:
# 你的逻辑
result = f"处理 {arg1} 并使用参数 {arg2}"
return result
except Exception as e:
return f"Error: {str(e)}"
# 注册工具
agent.tools.register(MyTool())
带外部 API 的工具
import httpx
from nanobot.agent.tools.base import Tool
class WeatherTool(Tool):
def __init__(self, api_key: str):
self.api_key = api_key
@property
def name(self) -> str:
return "get_weather"
@property
def description(self) -> str:
return "获取指定城市的天气信息"
@property
def parameters(self) -> dict[str, Any]:
return {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称(如:北京、上海)"
}
},
"required": ["city"]
}
async def execute(self, city: str) -> str:
try:
async with httpx.AsyncClient() as client:
response = await client.get(
f"https://api.example.com/weather",
params={"city": city, "key": self.api_key}
)
response.raise_for_status()
data = response.json()
return f"{city}的天气:{data['weather']}, 温度:{data['temp']}°C"
except httpx.HTTPError as e:
return f"无法获取天气信息:{str(e)}"
# 注册
agent.tools.register(WeatherTool(api_key="YOUR_API_KEY"))
技能开发模板
技能目录结构
~/.nanobot/skills/my-skill/
├── SKILL.md # 技能定义(必需)
├── script.py # 可选脚本
└── resources/ # 可选资源
└── data.json
SKILL.md 模板
---
name: my-skill
description: 技能的简短描述
available: true
---
# my-skill
完整的技能说明。
## 用途
这个技能用于...
## 使用方法
调用方式:
\```bash
python ~/.nanobot/skills/my-skill/script.py arg1 arg2
\```
## 示例
\```
用户: 请使用 my-skill 做某事
助手: [读取此文件] 好的,我会使用脚本...
\```
## 依赖
- Python 包: requests, beautifulsoup4
- 系统命令: curl
## 参数
- `arg1`: 第一个参数的说明
- `arg2`: 第二个参数的说明(可选)
技能脚本模板
#!/usr/bin/env python3
"""
my-skill 脚本
用法:
python script.py <arg1> <arg2>
"""
import sys
import json
def main():
if len(sys.argv) < 2:
print("Usage: script.py <arg1> [arg2]", file=sys.stderr)
sys.exit(1)
arg1 = sys.argv[1]
arg2 = sys.argv[2] if len(sys.argv) > 2 else None
# 你的逻辑
result = process(arg1, arg2)
# 输出结果(JSON 或纯文本)
print(json.dumps(result, ensure_ascii=False))
def process(arg1, arg2=None):
"""处理逻辑"""
return {"status": "success", "data": "..."}
if __name__ == "__main__":
main()
常见问题速查
Q: 如何修改默认模型?
A: 编辑 ~/.nanobot/config.json:
{
"agents": {
"defaults": {
"model": "anthropic/claude-sonnet-4-5"
}
}
}
Q: 如何启用 Web 搜索?
A: 添加 Brave Search API Key:
{
"tools": {
"web": {
"search": {
"apiKey": "BSA-YOUR_KEY"
}
}
}
}
Q: 如何查看详细日志?
A: 使用 --verbose 参数:
nanobot gateway --verbose
或在代码中设置日志级别:
from loguru import logger
logger.remove()
logger.add(sys.stderr, level="DEBUG")
Q: 工具返回值必须是字符串吗?
A: 是的。LLM 期望字符串格式的结果。如果你的工具返回复杂数据,可以使用 JSON:
async def execute(self, **kwargs) -> str:
result = {"status": "ok", "data": [...]}
return json.dumps(result, ensure_ascii=False)
Q: 如何限制工具执行时间?
A: 使用 asyncio.wait_for:
async def execute(self, **kwargs) -> str:
try:
result = await asyncio.wait_for(
self.long_running_task(**kwargs),
timeout=30.0
)
return result
except asyncio.TimeoutError:
return "操作超时"
Q: 会话保存在哪里?
A: 默认保存在 ~/.nanobot/sessions/ 目录,每个会话一个 JSON 文件。
Q: 如何清空会话历史?
A: 删除对应的会话文件:
rm ~/.nanobot/sessions/cli:default.json
Q: 可以同时使用多个 LLM 提供商吗?
A: 可以。配置多个提供商,然后在运行时指定:
provider1 = LiteLLMProvider(api_key="...", default_model="anthropic/...")
provider2 = LiteLLMProvider(api_key="...", default_model="openai/...")
agent1 = AgentLoop(bus, provider1, workspace, model="anthropic/claude-opus-4-5")
agent2 = AgentLoop(bus, provider2, workspace, model="openai/gpt-4")
Q: 如何禁用某个工具?
A: 不注册即可:
# 不调用 self.tools.register(UnwantedTool())
或者创建一个过滤的注册表:
class FilteredToolRegistry(ToolRegistry):
def register(self, tool: Tool):
if tool.name not in ["unwanted_tool"]:
super().register(tool)
Q: 技能和工具的区别是什么?
A:
- 工具(Tool):Python 代码实现的功能,Agent 可以直接调用
- 技能(Skill):Markdown 文档 + 可选脚本,Agent 通过
read_file读取后使用
技能更灵活,但需要 Agent 自己理解如何使用;工具更直接,但需要编写代码。
调试技巧
1. 查看 LLM 的原始请求和响应
from loguru import logger
# 在调用前
logger.debug(f"Messages to LLM: {json.dumps(messages, indent=2)}")
response = await provider.chat(messages, tools=...)
# 在调用后
logger.debug(f"LLM response: {response}")
2. 检查工具定义
definitions = agent.tools.get_definitions()
print(json.dumps(definitions, indent=2))
3. 手动测试工具
import asyncio
tool = MyTool()
result = asyncio.run(tool.execute(arg1="test"))
print(result)
4. 查看会话历史
session = agent.sessions.get_or_create("cli:default")
print(json.dumps(session.get_history(), indent=2))
性能优化建议
1. 减少 Token 使用
- 限制历史消息数量
- 精简系统提示
- 使用更小的模型
2. 加速工具执行
- 使用异步 HTTP 客户端(httpx)
- 并发执行独立的工具调用
- 添加缓存机制
3. 降低延迟
- 使用本地模型(vLLM)
- 减少工具数量(只注册需要的)
- 优化提示词长度
这份速查手册应该能帮助你快速找到需要的信息。如果需要更详细的解释,请参考完整的学习文档。 📖