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)
  • 减少工具数量(只注册需要的)
  • 优化提示词长度

这份速查手册应该能帮助你快速找到需要的信息。如果需要更详细的解释,请参考完整的学习文档。 📖