关键代码验证
技术研究 AI Agent 代码分析
nanobot 核心代码片段分析，验证其技术实现的简洁性和可扩展性
快速开始验证
安装与初始化
nanobot 的安装极为简单，体现了其”开箱即用”的设计理念：
# 方式一：从 PyPI 安装（稳定版）
pip install nanobot-ai
# 方式二：使用 uv 安装（更快）
uv tool install nanobot-ai
# 方式三：从源码安装（最新功能）
git clone https://github.com/HKUDS/nanobot.git
cd nanobot
pip install -e .初始化仅需一条命令：
nanobot onboard这会创建 ~/.nanobot/config.json 配置文件，并引导用户完成基本配置。
最小配置示例
{
"providers": {
"openrouter": {
"apiKey": "sk-or-v1-xxx"
}
},
"agents": {
"defaults": {
"model": "anthropic/claude-opus-4-5",
"provider": "openrouter"
}
}
}启动 Agent：
nanobot agent从安装到运行，仅需 2 分钟。
核心代码结构
项目目录结构
nanobot/
├── nanobot/
│ ├── agent/ # Agent 核心逻辑
│ │ ├── __init__.py
│ │ ├── core.py # 主循环
│ │ └── tools.py # 工具定义
│ ├── providers/ # LLM Provider
│ │ ├── __init__.py
│ │ ├── registry.py # Provider 注册表
│ │ └── base.py # 基类
│ ├── channels/ # 消息平台
│ │ ├── telegram.py
│ │ ├── discord.py
│ │ ├── whatsapp.py
│ │ └── ... # 其他平台
│ ├── memory/ # 记忆系统
│ │ └── store.py
│ ├── tools/ # 内置工具
│ │ ├── web_search.py
│ │ ├── memory.py
│ │ └── schedule.py
│ └── config/ # 配置管理
│ └── schema.py
├── bridge/ # WhatsApp Bridge (Go)
├── tests/
└── pyproject.toml核心代码量验证
运行项目提供的脚本：
bash core_agent_lines.sh输出示例：
Core agent lines: 3935这证实了 nanobot 的核心代码确实控制在约 4,000 行以内。
关键代码片段
1. Provider 注册机制
nanobot/providers/registry.py 的核心设计：
from dataclasses import dataclass
from typing import Tuple, Optional
@dataclass
class ProviderSpec:
"""Provider 规格定义"""
name: str # 配置字段名
keywords: Tuple[str, ...] # 模型名关键词
env_key: str # 环境变量名
display_name: str # 显示名称
litellm_prefix: Optional[str] = None # LiteLLM 前缀
skip_prefixes: Tuple[str, ...] = () # 跳过前缀
# Provider 注册表
PROVIDERS = {
"openai": ProviderSpec(
name="openai",
keywords=("gpt", "o1", "o3"),
env_key="OPENAI_API_KEY",
display_name="OpenAI",
litellm_prefix="openai",
),
"anthropic": ProviderSpec(
name="anthropic",
keywords=("claude",),
env_key="ANTHROPIC_API_KEY",
display_name="Anthropic",
litellm_prefix="anthropic",
),
"deepseek": ProviderSpec(
name="deepseek",
keywords=("deepseek",),
env_key="DEEPSEEK_API_KEY",
display_name="DeepSeek",
litellm_prefix="deepseek",
),
# ... 更多 Provider
}
def match_provider(model: str) -> Optional[ProviderSpec]:
"""根据模型名自动匹配 Provider"""
model_lower = model.lower()
for spec in PROVIDERS.values():
if any(kw in model_lower for kw in spec.keywords):
return spec
return None设计亮点：
- 使用
dataclass简化定义 - 注册表模式避免 if-elif 链
- 自动匹配降低配置复杂度
2. Agent 核心循环
nanobot/agent/core.py 的简化实现：
import asyncio
from typing import List, Dict, Any
class Agent:
def __init__(self, provider, tools: List[Tool]):
self.provider = provider
self.tools = {t.name: t for t in tools}
self.max_iterations = 10
async def run(self, messages: List[Dict]) -> str:
"""运行 Agent 循环"""
iteration = 0
while iteration < self.max_iterations:
# 调用 LLM
response = await self.provider.chat(
messages=messages,
tools=list(self.tools.values())
)
# 检查是否有工具调用
if not response.tool_calls:
return response.content
# 执行工具调用
tool_results = await asyncio.gather(*[
self.execute_tool(tc)
for tc in response.tool_calls
])
# 将结果加入消息历史
messages.append({
"role": "assistant",
"tool_calls": response.tool_calls
})
for result in tool_results:
messages.append({
"role": "tool",
"content": result
})
iteration += 1
return "达到最大迭代次数"
async def execute_tool(self, tool_call) -> str:
"""执行单个工具调用"""
tool = self.tools.get(tool_call.function.name)
if not tool:
return f"未知工具: {tool_call.function.name}"
args = json.loads(tool_call.function.arguments)
return await tool.execute(**args)设计亮点：
- 清晰的循环终止条件
- 工具并行执行（
asyncio.gather） - 迭代次数限制防止死循环
3. Channel 接口抽象
nanobot/channels/base.py：
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
@dataclass
class Message:
"""统一消息格式"""
content: str
sender_id: str
session_id: str
channel: str
attachments: Optional[List[Attachment]] = None
class BaseChannel(ABC):
"""Channel 基类"""
@abstractmethod
async def start(self):
"""启动 Channel"""
pass
@abstractmethod
async def stop(self):
"""停止 Channel"""
pass
@abstractmethod
async def send_message(
self,
session_id: str,
content: str,
**kwargs
):
"""发送消息"""
pass
async def handle_message(self, message: Message):
"""处理收到的消息（通用逻辑）"""
# 1. 权限检查
if not self.is_allowed(message.sender_id):
return
# 2. 转发给 Agent
response = await self.agent.run([
{"role": "user", "content": message.content}
])
# 3. 发送回复
await self.send_message(message.session_id, response)设计亮点：
- ABC 定义统一接口
Message数据类统一消息格式- 通用逻辑在基类实现，平台特定逻辑在子类
4. 添加新的 Channel
以 Telegram 为例：
# nanobot/channels/telegram.py
import telebot
from .base import BaseChannel, Message
class TelegramChannel(BaseChannel):
def __init__(self, config, agent):
self.bot = telebot.TeleBot(config.token)
self.agent = agent
self.allow_from = config.allow_from
async def start(self):
@self.bot.message_handler(func=lambda m: True)
def handle(message):
# 转换为统一消息格式
msg = Message(
content=message.text,
sender_id=str(message.from_user.id),
session_id=str(message.chat.id),
channel="telegram"
)
asyncio.create_task(self.handle_message(msg))
self.bot.infinity_polling()
async def send_message(self, session_id: str, content: str):
self.bot.send_message(int(session_id), content)
def is_allowed(self, sender_id: str) -> bool:
return sender_id in self.allow_from or "*" in self.allow_from新增 Channel 只需：
- 继承
BaseChannel - 实现三个抽象方法
- 在配置中启用
5. Memory 系统实现
nanobot/memory/store.py：
import json
from pathlib import Path
from typing import List, Dict
import subprocess
class MemoryStore:
"""基于文件的记忆存储"""
def __init__(self, base_path: Path):
self.memory_file = base_path / "memory.jsonl"
self.context_file = base_path / "context.json"
def add_memory(self, content: str, metadata: Dict = None):
"""添加记忆条目"""
entry = {
"content": content,
"metadata": metadata or {},
"timestamp": datetime.now().isoformat()
}
with open(self.memory_file, "a") as f:
f.write(json.dumps(entry) + "\n")
def search(self, query: str, limit: int = 5) -> List[Dict]:
"""搜索相关记忆（使用 grep）"""
# 使用 grep 进行文本搜索
result = subprocess.run(
["grep", "-i", query, str(self.memory_file)],
capture_output=True,
text=True
)
memories = []
for line in result.stdout.strip().split("\n")[:limit]:
if line:
memories.append(json.loads(line))
return memories
def get_context(self) -> Dict:
"""获取当前会话上下文"""
if self.context_file.exists():
return json.loads(self.context_file.read_text())
return {}
def save_context(self, context: Dict):
"""保存会话上下文"""
self.context_file.write_text(json.dumps(context, indent=2))设计亮点：
- 零依赖（仅使用标准库）
- JSONL 格式便于追加和检索
- grep 实现简单有效的关键词搜索
配置系统集成
配置 Schema
nanobot/config/schema.py 使用 Pydantic：
from pydantic import BaseModel
from typing import Optional, List, Dict
class ProviderConfig(BaseModel):
apiKey: Optional[str] = None
apiBase: Optional[str] = None
class ProvidersConfig(BaseModel):
openrouter: ProviderConfig = ProviderConfig()
openai: ProviderConfig = ProviderConfig()
anthropic: ProviderConfig = ProviderConfig()
deepseek: ProviderConfig = ProviderConfig()
# ... 其他 Provider
class TelegramConfig(BaseModel):
enabled: bool = False
token: Optional[str] = None
allowFrom: List[str] = ["*"]
class ChannelsConfig(BaseModel):
telegram: TelegramConfig = TelegramConfig()
discord: DiscordConfig = DiscordConfig()
# ... 其他 Channel
class AgentConfig(BaseModel):
model: str = "gpt-4"
provider: Optional[str] = None
maxIterations: int = 10
class Config(BaseModel):
providers: ProvidersConfig = ProvidersConfig()
channels: ChannelsConfig = ChannelsConfig()
agents: Dict[str, AgentConfig] = {"defaults": AgentConfig()}优势：
- 自动验证配置格式
- 默认值减少必填项
- 类型安全
扩展示例：添加新工具
定义工具
from nanobot.tools import Tool, ToolResult
class WeatherTool(Tool):
name = "get_weather"
description = "获取指定城市的天气信息"
parameters = {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
}
},
"required": ["city"]
}
async def execute(self, city: str) -> ToolResult:
# 调用天气 API
weather_data = await self.fetch_weather(city)
return ToolResult(
success=True,
content=f"{city} 当前天气：{weather_data}"
)注册工具
# 在 nanobot/tools/__init__.py 中注册
TOOLS = [
WebSearchTool(),
MemoryTool(),
ScheduleTool(),
WeatherTool(), # 新增
]工具会自动出现在 LLM 的工具列表中，无需额外配置。