1. nanobot智能体技能与工具调用机制概述
在AI智能体开发领域,如何实现功能扩展和外部交互一直是核心挑战。nanobot智能体通过创新的技能(Skills)系统和工具调用(Tool Call)机制,构建了一套完整的解决方案。这套系统不仅支持模块化功能扩展,还能与外部环境进行安全、高效的交互。
1.1 核心设计理念
nanobot的设计遵循三个基本原则:
- 模块化:将不同功能拆分为独立技能包,每个技能包含完整的元数据和使用说明
- 渐进式加载:根据实际需求动态加载技能内容,优化上下文窗口使用效率
- 标准化交互:采用OpenAI Function Calling标准实现工具调用,确保兼容性和扩展性
这种架构使得智能体能够:
- 通过技能系统快速扩展领域知识
- 通过工具调用安全执行外部操作
- 保持核心系统的简洁性和稳定性
提示:在实际开发中,模块化设计可以显著降低系统耦合度,使不同功能能够独立开发和更新。
2. 技能(Skills)机制详解
2.1 技能定义与结构
技能是扩展智能体能力的知识包,采用Markdown文件存储,包含YAML前置元数据和详细指令内容。一个完整的技能目录结构如下:
code复制github/ # 技能名称目录
├── SKILL.md # 主技能文件(必需)
│ ├── YAML frontmatter # 元数据部分
│ └── Markdown内容 # 详细指令
└── resources/ # 资源目录(可选)
├── scripts/ # 可执行脚本
├── references/ # 参考文档
└── templates/ # 模板文件
2.1.1 SKILL.md文件示例
markdown复制---
name: github
description: "使用gh CLI与GitHub交互。支持issue、PR、CI运行和高级查询。"
metadata: {
"nanobot": {
"emoji": "🐙",
"requires": {"bins": ["gh"]},
"always": false
}
}
---
# GitHub技能指南
使用`gh`命令行工具与GitHub交互。不在git目录时需指定`--repo owner/repo`参数。
## Pull Requests操作
检查PR的CI状态:
```bash
gh pr checks 55 --repo owner/repo
列出最近的工作流运行:
bash复制gh run list --repo owner/repo --limit 10
Issues管理
创建新issue:
bash复制gh issue create --title "Bug报告" --body "详细描述..."
搜索issues:
bash复制gh search issues "label:bug is:open"
API高级查询
使用GraphQL查询:
bash复制gh api graphql -f query='
query {
repository(owner:"owner", name:"repo") {
issues(last:5) {
nodes {
title
url
}
}
}
}
'
注意:使用API查询需要配置GITHUB_TOKEN环境变量
code复制
#### 2.1.2 元数据字段说明
| 字段 | 说明 | 示例 |
|------|------|------|
| `name` | 技能唯一标识符 | `github`, `docker` |
| `description` | 技能描述(LLM触发依据) | "使用gh CLI与GitHub交互" |
| `metadata.nanobot.emoji` | 技能图标 | `"🐙"` |
| `metadata.nanobot.requires.bins` | 依赖的CLI工具 | `["gh", "jq"]` |
| `metadata.nanobot.requires.env` | 依赖的环境变量 | `["GITHUB_TOKEN"]` |
| `metadata.nanobot.always` | 是否始终加载完整内容 | `false` |
### 2.2 技能加载系统
技能加载器(`SkillsLoader`)负责技能的发现、加载和依赖检查,核心功能包括:
#### 2.2.1 技能发现优先级
1. **工作区技能**:`workspace/skills/`目录下的用户自定义技能
2. **内置技能**:`nanobot/skills/`目录下的系统预置技能
3. **依赖过滤**:自动检查并过滤掉依赖不满足的技能
```python
def list_skills(self, filter_unavailable=True) -> list[dict]:
skills = []
# 1. 加载工作区技能
if self.workspace_skills.exists():
for skill_dir in self.workspace_skills.iterdir():
skill_file = skill_dir / "SKILL.md"
if skill_file.exists():
skills.append({
"name": skill_dir.name,
"path": str(skill_file),
"source": "workspace"
})
# 2. 加载内置技能(不覆盖工作区同名技能)
if self.builtin_skills.exists():
for skill_dir in self.builtin_skills.iterdir():
skill_file = skill_dir / "SKILL.md"
if skill_file.exists() and not any(
s["name"] == skill_dir.name for s in skills
):
skills.append({
"name": skill_dir.name,
"path": str(skill_file),
"source": "builtin"
})
# 3. 按需过滤不可用技能
return [s for s in skills if not filter_unavailable or self._check_requirements(s)]
2.2.2 依赖检查机制
python复制def _check_requirements(self, skill_meta: dict) -> bool:
requires = skill_meta.get("metadata", {}).get("nanobot", {}).get("requires", {})
# 检查CLI工具依赖
for bin_name in requires.get("bins", []):
if not shutil.which(bin_name):
return False
# 检查环境变量依赖
for env_var in requires.get("env", []):
if not os.environ.get(env_var):
return False
return True
2.3 渐进式加载设计
为优化上下文窗口使用,nanobot采用三级加载策略:
- Level 1 - 元数据摘要:始终加载,包含技能名称、描述和可用状态
- Level 2 - 完整技能内容:按需通过
read_file工具加载SKILL.md - Level 3 - 资源文件:按需加载scripts、references等附加资源
2.3.1 XML摘要生成
python复制def build_skills_summary(self) -> str:
skills = self.list_skills(filter_unavailable=False)
if not skills:
return ""
xml_lines = ["<skills>"]
for skill in skills:
meta = self._get_skill_metadata(skill["name"])
available = self._check_requirements(meta)
xml_lines.append(f' <skill available="{str(available).lower()}">')
xml_lines.append(f" <name>{escape_xml(skill['name'])}</name>")
xml_lines.append(f" <description>{escape_xml(meta['description'])}</description>")
xml_lines.append(f" <location>{escape_xml(skill['path'])}</location>")
if not available:
missing = self._get_missing_requirements(meta)
if missing:
xml_lines.append(f" <requires>{escape_xml(missing)}</requires>")
xml_lines.append(" </skill>")
xml_lines.append("</skills>")
return "\n".join(xml_lines)
2.3.2 工作流程示例
- LLM读取技能摘要,发现需要
github技能 - 调用
read_file工具加载/path/to/github/SKILL.md - 解析技能内容,执行相关
gh命令 - 如需额外资源,再按需加载
scripts/或references/中的文件
实操心得:渐进式加载能显著减少上下文占用,但要注意技能描述的清晰度,确保LLM能准确判断何时需要加载完整技能。
3. 工具调用(Tool Call)机制
3.1 工具定义与注册
3.1.1 Tool抽象基类
所有工具必须实现以下抽象方法:
python复制class Tool(ABC):
@property
@abstractmethod
def name(self) -> str:
"""工具唯一标识"""
@property
@abstractmethod
def description(self) -> str:
"""工具功能描述"""
@property
@abstractmethod
def parameters(self) -> dict:
"""参数JSON Schema定义"""
@abstractmethod
async def execute(self, **kwargs) -> str:
"""异步执行工具"""
3.1.2 示例:ExecTool实现
python复制class ExecTool(Tool):
def __init__(self, timeout=60, restrict_to_workspace=True):
self.timeout = timeout
self.restrict_to_workspace = restrict_to_workspace
@property
def name(self) -> str:
return "exec"
@property
def description(self) -> str:
return "执行shell命令并返回输出。使用时需谨慎。"
@property
def parameters(self) -> dict:
return {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "要执行的shell命令"
},
"working_dir": {
"type": "string",
"description": "命令执行的工作目录"
}
},
"required": ["command"]
}
async def execute(self, command: str, working_dir: str = None, **kwargs) -> str:
# 安全检查
if self.restrict_to_workspace:
if working_dir and not working_dir.startswith(self.workspace):
return "错误:工作目录必须在工作区内"
if any(blocked in command for blocked in ["rm -rf", "chmod", "sudo"]):
return "错误:禁止执行危险命令"
# 执行命令
process = await asyncio.create_subprocess_shell(
command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=working_dir,
)
try:
stdout, stderr = await asyncio.wait_for(
process.communicate(),
timeout=self.timeout
)
except asyncio.TimeoutError:
process.kill()
return "错误:命令执行超时"
# 组装结果
output = []
if stdout:
output.append(stdout.decode("utf-8", errors="replace"))
if stderr:
output.append(f"STDERR:\n{stderr.decode('utf-8', errors='replace')}")
return "\n".join(output) if output else "(无输出)"
3.2 工具调用流程
3.2.1 完整工作流
- LLM请求:智能体将用户消息和工具定义发送给LLM
- 工具选择:LLM决定是否使用工具及使用哪些工具
- 参数生成:LLM根据工具schema生成合规参数
- 工具执行:系统执行工具并返回结果
- 结果处理:LLM基于工具结果生成最终响应
3.2.2 代码实现
python复制async def _process_message(self, msg: InboundMessage) -> OutboundMessage:
messages = self.context.build_messages(msg)
final_content = None
for _ in range(self.max_iterations):
# 调用LLM
response = await self.provider.chat(
messages=messages,
tools=self.tools.get_definitions(),
model=self.model
)
if response.has_tool_calls:
# 添加assistant消息到历史
messages = self.context.add_assistant_message(
messages,
response.content,
[tc.to_dict() for tc in response.tool_calls]
)
# 执行每个工具调用
for tool_call in response.tool_calls:
result = await self.tools.execute(
tool_call.name,
tool_call.arguments
)
# 添加工具结果到历史
messages = self.context.add_tool_result(
messages,
tool_call.id,
tool_call.name,
result
)
else:
final_content = response.content
break
return OutboundMessage(
channel=msg.channel,
chat_id=msg.chat_id,
content=final_content
)
3.3 消息格式规范
3.3.1 标准消息序列
- 用户消息:
json复制{
"role": "user",
"content": "请列出当前目录文件"
}
- Assistant工具调用:
json复制{
"role": "assistant",
"content": "正在为您列出目录内容...",
"tool_calls": [
{
"id": "call_123",
"type": "function",
"function": {
"name": "list_dir",
"arguments": "{\"path\":\".\"}"
}
}
]
}
- 工具结果:
json复制{
"role": "tool",
"tool_call_id": "call_123",
"name": "list_dir",
"content": "README.md\nsrc/\nmain.py"
}
- Assistant最终回复:
json复制{
"role": "assistant",
"content": "当前目录包含:\n- README.md\n- src/\n- main.py"
}
3.3.2 消息管理实现
python复制def add_assistant_message(self, messages, content, tool_calls=None):
msg = {"role": "assistant", "content": content or ""}
if tool_calls:
msg["tool_calls"] = tool_calls
messages.append(msg)
return messages
def add_tool_result(self, messages, tool_call_id, tool_name, result):
messages.append({
"role": "tool",
"tool_call_id": tool_call_id,
"name": tool_name,
"content": str(result)[:4096] # 限制长度
})
return messages
4. 系统集成与最佳实践
4.1 技能与工具协同工作
技能和工具在nanobot中形成互补关系:
- 技能提供领域知识和使用指导
- 工具提供实际执行能力
- LLM作为协调中心,根据技能指导选择合适的工具
4.1.1 典型工作流
- 用户请求需要特定领域知识
- LLM检查技能摘要,识别相关技能
- 加载技能完整内容,获取详细指导
- 根据技能说明选择并调用适当工具
- 解析工具结果,生成用户响应
4.2 性能优化技巧
-
技能设计:
- 保持技能描述精准,提高LLM识别准确率
- 复杂技能拆分为多个小技能
- 常用技能标记为
always=true减少加载延迟
-
工具实现:
- 工具执行应快速返回,长时间操作需异步实现
- 参数验证前置,尽早返回错误
- 结果长度控制在合理范围内(如4K字符)
-
系统配置:
- 根据使用场景调整
max_iterations(默认5-10次) - 为不同工具设置合理超时(CLI命令30-60秒)
- 启用结果缓存减少重复计算
- 根据使用场景调整
4.3 安全注意事项
-
技能安全:
- 严格审核第三方技能内容
- 技能目录权限设置为只读
- 危险操作(如文件删除)需明确警告
-
工具安全:
- CLI命令执行需过滤危险操作
- 文件操作限制在工作目录内
- 敏感操作需二次确认
-
系统安全:
- 定期检查工具依赖版本
- 监控异常工具调用模式
- 实现操作审计日志
避坑指南:在实际部署中,务必对
exec工具进行严格限制,避免命令注入风险。建议使用白名单机制限制可执行命令范围。
5. 扩展与定制开发
5.1 开发自定义技能
创建新技能的推荐步骤:
- 在
workspace/skills/下创建技能目录 - 编写
SKILL.md文件,包含完整元数据和指南 - 测试技能加载和依赖检查
- 验证LLM能正确识别和使用技能
5.1.1 示例:开发天气查询技能
- 创建目录结构:
code复制weather/
├── SKILL.md
└── resources/
├── scripts/
│ └── fetch_weather.py
└── templates/
└── weather_report.md
- 编写
SKILL.md:
markdown复制---
name: weather
description: "查询当前天气和预报。需要城市名称或坐标。"
metadata: {
"nanobot": {
"emoji": "🌤️",
"requires": {
"bins": ["python3"],
"env": ["WEATHER_API_KEY"]
}
}
}
---
# 天气查询技能
使用OpenWeatherMap API查询天气信息。
## 基本用法
查询当前天气:
```bash
python3 scripts/fetch_weather.py --city "北京"
查询天气预报:
bash复制python3 scripts/fetch_weather.py --city "上海" --forecast
输出格式
结果包含:
- 当前温度(℃)
- 天气状况(晴/雨等)
- 湿度(%)
- 风速(m/s)
- 日出/日落时间
code复制
### 5.2 开发自定义工具
创建新工具的推荐步骤:
1. 继承`Tool`基类实现核心方法
2. 编写详细的参数schema
3. 实现安全的`execute`方法
4. 注册工具到系统
#### 5.2.1 示例:数据库查询工具
```python
class DatabaseQueryTool(Tool):
def __init__(self, db_url: str, timeout: int = 30):
self.db_url = db_url
self.timeout = timeout
@property
def name(self) -> str:
return "db_query"
@property
def description(self) -> str:
return "执行安全的SQL查询并返回结果。仅支持SELECT查询。"
@property
def parameters(self) -> dict:
return {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "要执行的SELECT查询语句"
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 100,
"default": 10,
"description": "返回结果的最大行数"
}
},
"required": ["query"]
}
async def execute(self, query: str, limit: int = 10, **kwargs) -> str:
# 安全检查
if not query.strip().upper().startswith("SELECT"):
return "错误:只允许执行SELECT查询"
if ";" in query:
return "错误:查询不能包含分号"
# 执行查询
try:
async with async_timeout.timeout(self.timeout):
async with aiopg.connect(self.db_url) as conn:
async with conn.cursor() as cur:
await cur.execute(f"{query} LIMIT {limit}")
rows = await cur.fetchall()
return "\n".join(str(row) for row in rows)
except Exception as e:
return f"查询错误:{str(e)}"
5.3 调试技巧
-
技能调试:
- 使用
skills_summary检查技能是否被正确加载 - 验证依赖检查逻辑
- 测试技能内容加载速度
- 使用
-
工具调试:
- 记录工具调用参数和结果
- 验证参数schema的准确性
- 测试边界条件和错误处理
-
集成调试:
- 检查消息历史完整性
- 监控上下文窗口使用情况
- 分析LLM决策过程
调试心得:在实际开发中,建议为每个工具实现详细的日志记录,包括入参、执行时间和结果摘要。这对排查复杂问题非常有帮助。
6. 架构设计与实现细节
6.1 核心组件交互
nanobot系统的核心组件及其交互关系:
code复制+-------------+ +----------------+ +------------+
| SkillLoader | | ToolRegistry | | LLMProvider |
+-------------+ +----------------+ +------------+
| | |
v v v
+------------------------------------------------+
| ContextBuilder |
+------------------------------------------------+
|
v
+------------------------------------------------+
| AgentLoop |
+------------------------------------------------+
|
v
+-------------+ +----------------+
| Session | | MessageBus |
+-------------+ +----------------+
6.2 关键数据结构
6.2.1 技能元数据结构
python复制class SkillMetadata(TypedDict):
name: str
description: str
metadata: dict
path: str
source: Literal["workspace", "builtin"]
available: bool
6.2.2 工具调用请求
python复制@dataclass
class ToolCallRequest:
id: str
name: str
arguments: dict[str, Any]
6.2.3 LLM响应
python复制@dataclass
class LLMResponse:
content: str | None
tool_calls: list[ToolCallRequest]
finish_reason: str
usage: dict[str, int]
reasoning_content: str | None
6.3 性能考量
-
上下文管理:
- 技能采用渐进式加载减少token消耗
- 工具结果进行适当摘要
- 历史消息智能截断
-
并发处理:
- 工具执行全异步化
- 并行执行独立工具调用
- 实现请求超时和取消
-
缓存策略:
- 缓存常用技能内容
- 缓存工具计算结果
- 实现智能缓存失效
7. 实际应用案例
7.1 GitHub仓库管理自动化
场景:自动处理GitHub issue和PR
实现步骤:
- 加载
github技能 - 识别用户请求类型(issue/PR)
- 调用
exec工具执行相应gh命令 - 解析结果并生成用户报告
示例对话:
code复制用户:请查看PR #123的CI状态
智能体:正在检查PR #123的状态...
(调用 gh pr checks 123 --repo owner/repo)
PR #123的CI状态:
- build: 通过 ✅
- test: 运行中 ⏳
- lint: 失败 ❌
7.2 数据分析工作流
场景:执行SQL查询并可视化结果
实现步骤:
- 加载
sql和visualization技能 - 解析用户查询意图
- 调用
db_query工具执行查询 - 调用
plot工具生成图表 - 返回图表和数据分析
示例对话:
code复制用户:显示最近一个月销售额趋势
智能体:正在查询销售数据...
(调用 db_query "SELECT date, amount FROM sales WHERE date > NOW() - INTERVAL '1 month' ORDER BY date")
已生成销售额趋势图:
[显示折线图]
过去30天日均销售额:$12,345
峰值出现在2023-11-15:$18,900
7.3 系统监控与告警
场景:服务器状态监控
实现步骤:
- 加载
monitoring技能 - 定期调用
exec执行监控命令 - 分析结果判断异常
- 调用
message工具发送告警
示例实现:
python复制class MonitoringTool(Tool):
async def execute(self, **kwargs) -> str:
# 检查CPU使用率
cpu = await self._exec_command("top -bn1 | grep 'Cpu(s)'")
if float(cpu.split(",")[0].replace("%Cpu(s):", "")) > 90:
await self.message_tool.execute(
recipient="admin",
content="警告:CPU使用率超过90%"
)
# 检查磁盘空间
disk = await self._exec_command("df -h / | awk 'NR==2{print $5}'")
if int(disk.replace("%", "")) > 85:
await self.message_tool.execute(
recipient="admin",
content="警告:根分区空间不足"
)
return "监控检查完成"
8. 常见问题与解决方案
8.1 技能加载问题
问题1:技能未被正确识别
- 检查技能目录结构是否符合规范
- 验证
SKILL.md的YAML元数据格式 - 确认技能位于正确目录(
workspace/skills/或nanobot/skills/)
问题2:依赖检查误报
- 手动验证
which <command>和echo $VAR结果 - 检查技能元数据中的
requires定义 - 确认执行环境与测试环境一致
8.2 工具调用问题
问题1:LLM未调用预期工具
- 检查工具
description是否清晰描述功能 - 验证参数
schema是否准确 - 确保工具���正确注册到
ToolRegistry
问题2:工具执行失败
- 检查工具日志确认输入参数
- 验证依赖环境和权限
- 测试独立执行工具逻辑
8.3 性能优化问题
问题1:响应延迟高
- 分析消息历史长度
- 检查工具执行时间
- 评估LLM响应时间
问题2:上下文窗口溢出
- 启用渐进式加载
- 优化技能内容简洁性
- 实现历史消息摘要
9. 未来发展展望
虽然nanobot的技能和工具系统已经相当完善,但在实际使用中我们发现几个值得深入探索的方向:
-
技能市场:建立共享技能仓库,支持一键安装社区贡献的技能包。这需要解决技能的安全审核和依赖管理问题。
-
动态工具组合:允许工具在运行时动态组合,形成更复杂的操作流程。例如将
db_query和chart工具自动串联生成数据报表。 -
技能版本管理:为技能添加版本控制和更新机制,确保系统稳定性和向后兼容性。
-
可视化编排:提供图形界面用于编排技能和工具的工作流,降低非技术用户的使用门槛。
-
性能分析工具:内置性能分析功能,帮助开发者识别系统瓶颈和优化机会。
在实际项目中,我们正在尝试将这些想法逐步实现。特别是技能市场的概念,已经在小范围内测试了技能共享机制,用户反馈非常积极。
