nanobot-技能与工具调用机制详解
nanobot智能体采用模块化技能系统与工具调用机制实现功能扩展。技能通过标准Markdown文件定义,包含YAML元数据和详细指令,支持渐进式加载策略。工具调用遵循OpenAI标准格式,包含完整的注册、模式生成、执行循环流程。系统优先加载工作区技能,自动检查依赖条件,并生成XML摘要供智能体按需调用。核心实现位于skills.py和tool_call.py,提供从技能发现到工具执行的完整工作流。
nanobot 智能体技能与工具调用机制详解
目录
技能(Skills)机制
技能定义与格式
技能是扩展智能体能力的模块化知识包,通过 Markdown 文件存储,指导智能体如何使用特定工具或执行特定任务。
技能目录结构
skill-name/
├── SKILL.md (必需)
│ ├── YAML frontmatter
│ │ ├── name: 技能名称
│ │ ├── description: 技能描述(触发机制)
│ │ └── metadata: nanobot专用元数据
│ └── Markdown 指令体
└── 可选资源目录
├── scripts/ # 可执行代码
├── references/ # 参考文档
└── assets/ # 模板/图标等资源
SKILL.md 示例 (GitHub技能)
---
name: github
description: "Interact with GitHub using the `gh` CLI. Use `gh issue`, `gh pr`, `gh run`, and `gh api` for issues, PRs, CI runs, and advanced queries."
metadata: {
"nanobot": {
"emoji": "🐙",
"requires": {"bins": ["gh"]},
"always": false
}
}
---
# GitHub Skill
Use the `gh` CLI to interact with GitHub. Always specify `--repo owner/repo` when not in a git directory, or use URLs directly.
## Pull Requests
Check CI status on a PR:
gh pr checks 55 --repo owner/repo
List recent workflow runs:
gh run list --repo owner/repo --limit 10
前置元数据字段
| 字段 | 说明 | 示例 |
|---|---|---|
name |
技能标识符(目录名) | github, weather |
description |
技能描述(LLM触发依据) | “获取当前天气和预报” |
metadata.nanobot.emoji |
技能图标 | "🌤️" |
metadata.nanobot.requires.bins |
依赖的CLI工具 | ["gh", "curl"] |
metadata.nanobot.requires.env |
依赖的环境变量 | ["GITHUB_TOKEN"] |
metadata.nanobot.always |
是否总是加载完整内容 | true/false |
技能加载系统
SkillsLoader 核心功能
位置: nanobot/agent/skills.py:13-228
class SkillsLoader:
def __init__(self, workspace: Path, builtin_skills_dir: Path | None = None):
self.workspace = workspace
self.workspace_skills = workspace / "skills"
self.builtin_skills = builtin_skills_dir or BUILTIN_SKILLS_DIR
技能发现与加载优先级
def list_skills(self, filter_unavailable: bool = True) -> list[dict[str, str]]:
"""
列出所有可用技能
优先级: workspace/skills/ > nanobot/skills/
"""
skills = []
# 1. Workspace技能(最高优先级)
if self.workspace_skills.exists():
for skill_dir in self.workspace_skills.iterdir():
if skill_dir.is_dir():
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 and self.builtin_skills.exists():
for skill_dir in self.builtin_skills.iterdir():
if skill_dir.is_dir():
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. 按依赖过滤
if filter_unavailable:
return [s for s in skills if self._check_requirements(s)]
return skills
依赖检查机制
def _check_requirements(self, skill_meta: dict) -> bool:
"""检查技能依赖是否满足"""
requires = skill_meta.get("requires", {})
# 检查CLI工具
for b in requires.get("bins", []):
if not shutil.which(b):
return False
# 检查环境变量
for env in requires.get("env", []):
if not os.environ.get(env):
return False
return True
技能内容加载
def load_skill(self, name: str) -> str | None:
"""按名称加载技能完整内容"""
# 先查workspace
workspace_skill = self.workspace_skills / name / "SKILL.md"
if workspace_skill.exists():
return workspace_skill.read_text(encoding="utf-8")
# 再查内置技能
if self.builtin_skills:
builtin_skill = self.builtin_skills / name / "SKILL.md"
if builtin_skill.exists():
return builtin_skill.read_text(encoding="utf-8")
return None
构建技能摘要(用于渐进加载)
def build_skills_summary(self) -> str:
"""
构建所有技能的XML格式摘要
用于渐进加载 - LLM可用read_file按需加载完整内容
"""
all_skills = self.list_skills(filter_unavailable=False)
if not all_skills:
return ""
lines = ["<skills>"]
for s in all_skills:
name = escape_xml(s["name"])
desc = escape_xml(self._get_skill_description(s["name"]))
skill_meta = self._get_skill_meta(s["name"])
available = self._check_requirements(skill_meta)
lines.append(f' <skill available="{str(available).lower()}">')
lines.append(f" <name>{name}</name>")
lines.append(f" <description>{desc}</description>")
lines.append(f" <location>{s['path']}</location>")
# 显示不可用技能的缺失依赖
if not available:
missing = self._get_missing_requirements(skill_meta)
if missing:
lines.append(f" <requires>{escape_xml(missing)}</requires>")
lines.append(f" </skill>")
lines.append("</skills>")
return "\n".join(lines)
技能集成到智能体
系统提示词构建流程
位置: nanobot/agent/context.py:28-71
def build_system_prompt(self, skill_names: list[str] | None = None) -> str:
"""从bootstrap文件、内存、技能构建系统提示词"""
parts = []
# 1. 核心身份
parts.append(self._get_identity())
# 2. Bootstrap文件
bootstrap = self._load_bootstrap_files() # AGENTS.md, SOUL.md, USER.md等
if bootstrap:
parts.append(bootstrap)
# 3. 内存上下文
memory = self.memory.get_memory_context()
if memory:
parts.append(f"# Memory\n\n{memory}")
# 4. 技能 - 渐进式加载
# 4.1 总是加载的技能: 包含完整内容
always_skills = self.skills.get_always_skills() # metadata.always=true
if always_skills:
always_content = self.skills.load_skills_for_context(always_skills)
if always_content:
parts.append(f"# Active Skills\n\n{always_content}")
# 4.2 可用技能: 只显示摘要(LLM用read_file按需加载)
skills_summary = self.skills.build_skills_summary()
if skills_summary:
parts.append(f"""# Skills
The following skills extend your capabilities. To use a skill, read its SKILL.md file using the read_file tool.
Skills with available="false" need dependencies installed first.
{skills_summary}""")
return "\n\n---\n\n".join(parts)
技能加载决策
def get_always_skills(self) -> list[str]:
"""获取标记为always=true且依赖满足的技能"""
result = []
for s in self.list_skills(filter_unavailable=True):
meta = self.get_skill_metadata(s["name"]) or {}
skill_meta = self._parse_nanobot_metadata(meta.get("metadata", ""))
if skill_meta.get("always") or meta.get("always"):
result.append(s["name"])
return result
渐进式加载设计
技能使用三级加载机制以高效管理上下文窗口:
Level 1: 元数据(始终在上下文)
- 内容: 技能名称 + 描述 + 可用性状态
- 大小: ~100词/技能
- 作用: LLM决策是否使用该技能
- 格式: XML摘要
<skills>
<skill available="true">
<name>github</name>
<description>Interact with GitHub using the gh CLI</description>
<location>/path/to/github/SKILL.md</location>
</skill>
<skill available="false">
<name>docker</name>
<description>Manage Docker containers and images</description>
<requires>CLI: docker</requires>
</skill>
</skills>
Level 2: SKILL.md 主体(技能触发时加载)
- 内容: 完整的技能指令
- 大小: <5k词(推荐)
- 触发: LLM调用
read_file工具读取SKILL.md - 示例: GitHub技能的完整使用说明
Level 3: 捆绑资源(按需加载)
- 内容: scripts/, references/, assets/
- 大小: 无限制(脚本可直接执行)
- 触发: LLM根据需要读取或执行
- 示例:
scripts/rotate_pdf.py可直接执行不读入上下文
工作流程示例
1. LLM看到技能摘要中的description
↓
2. 判断需要github技能处理用户请求
↓
3. 调用read_file工具读取/path/to/github/SKILL.md
↓
4. 获得技能完整内容后执行gh命令
↓
5. 如需要,读取scripts/或references/中的资源
工具调用(Tool Call)机制
Tool 定义与注册
Tool 抽象基类
位置: nanobot/agent/tools/base.py:7-103
class Tool(ABC):
"""工具抽象基类"""
@property
@abstractmethod
def name(self) -> str:
"""工具名称,用于函数调用"""
pass
@property
@abstractmethod
def description(self) -> str:
"""工具描述,LLM通过此了解工具用途"""
pass
@property
@abstractmethod
def parameters(self) -> dict[str, Any]:
"""JSON Schema格式的参数定义"""
pass
@abstractmethod
async def execute(self, **kwargs: Any) -> str:
"""执行工具并返回字符串结果"""
pass
工具示例: ExecTool(Shell执行)
位置: nanobot/agent/tools/shell.py:12-142
class ExecTool(Tool):
"""Shell命令执行工具"""
def __init__(
self,
timeout: int = 60,
working_dir: str | None = None,
restrict_to_workspace: bool = False,
):
self.timeout = timeout
self.working_dir = working_dir
self.restrict_to_workspace = restrict_to_workspace
@property
def name(self) -> str:
return "exec"
@property
def description(self) -> str:
return "Execute a shell command and return its output. Use with caution."
@property
def parameters(self) -> dict[str, Any]:
return {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "The shell command to execute"
},
"working_dir": {
"type": "string",
"description": "Optional working directory for the command"
}
},
"required": ["command"]
}
async def execute(self, command: str, working_dir: str | None = None, **kwargs: Any) -> str:
# 安全检查
guard_error = self._guard_command(command, cwd)
if guard_error:
return guard_error
# 执行命令
process = await asyncio.create_subprocess_shell(
command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=cwd,
)
stdout, stderr = await asyncio.wait_for(
process.communicate(),
timeout=self.timeout
)
# 返回结果
output_parts = []
if stdout:
output_parts.append(stdout.decode("utf-8", errors="replace"))
if stderr:
output_parts.append(f"STDERR:\n{stderr.decode('utf-8', errors='replace')}")
return "\n".join(output_parts) if output_parts else "(no output)"
文件系统工具示例: ReadFileTool
位置: nanobot/agent/tools/filesystem.py:17-57
class ReadFileTool(Tool):
"""文件读取工具"""
def __init__(self, allowed_dir: Path | None = None):
self._allowed_dir = allowed_dir
@property
def name(self) -> str:
return "read_file"
@property
def description(self) -> str:
return "Read the contents of a file at given path."
@property
def parameters(self) -> dict[str, Any]:
return {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "The file path to read"
}
},
"required": ["path"]
}
async def execute(self, path: str, **kwargs: Any) -> str:
try:
file_path = _resolve_path(path, self._allowed_dir)
if not file_path.exists():
return f"Error: File not found: {path}"
if not file_path.is_file():
return f"Error: Not a file: {path}"
content = file_path.read_text(encoding="utf-8")
return content
except Exception as e:
return f"Error reading file: {str(e)}"
Tool Schema 生成
转换为 OpenAI Function Calling 格式
位置: nanobot/agent/tools/base.py:93-102
def to_schema(self) -> dict[str, Any]:
"""转换工具为OpenAI function schema格式"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters # JSON Schema
}
}
JSON Schema 参数定义示例
# ExecTool 的参数定义
{
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "The shell command to execute"
},
"working_dir": {
"type": "string",
"description": "Optional working directory for the command"
}
},
"required": ["command"]
}
LLM 调用传递 Tools
AgentLoop 调用 LLM
位置: nanobot/agent/loop.py:195-199
async def _process_message(self, msg: InboundMessage) -> OutboundMessage | None:
# ...构建消息上下文...
# Agent循环
while iteration < self.max_iterations:
# 调用LLM,传入工具定义
response = await self.provider.chat(
messages=messages,
tools=self.tools.get_definitions(), # 传递所有工具的schema列表
model=self.model
)
# 处理响应...
ToolRegistry 获取工具定义
位置: nanobot/agent/tools/registry.py:34-36
def get_definitions(self) -> list[dict[str, Any]]:
"""获取所有已注册工具的OpenAI格式定义"""
return [tool.to_schema() for tool in self._tools.values()]
LiteLLMProvider 传递给底层 API
位置: nanobot/providers/litellm_provider.py:143-145
async def chat(
self,
messages: list[dict[str, Any]],
tools: list[dict[str, Any]] | None = None,
model: str | None = None,
...
) -> LLMResponse:
"""通过LiteLLM发送聊天请求"""
kwargs = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature,
}
# 传递工具定义给LLM
if tools:
kwargs["tools"] = tools
kwargs["tool_choice"] = "auto" # 让LLM自动决定是否使用工具
response = await acompletion(**kwargs)
return self._parse_response(response)
LLM 响应解析
解析 Tool Calls
位置: nanobot/providers/litellm_provider.py:157-195
def _parse_response(self, response: Any) -> LLMResponse:
"""解析LiteLLM响应为标准格式"""
choice = response.choices[0]
message = choice.message
tool_calls = []
if hasattr(message, "tool_calls") and message.tool_calls:
for tc in message.tool_calls:
# 解析参数(JSON字符串)
args = tc.function.arguments
if isinstance(args, str):
try:
args = json.loads(args)
except json.JSONDecodeError:
args = {"raw": args}
# 构建工具调用请求
tool_calls.append(ToolCallRequest(
id=tc.id,
name=tc.function.name,
arguments=args
))
return LLMResponse(
content=message.content,
tool_calls=tool_calls, # 包含所有工具调用
finish_reason=choice.finish_reason or "stop",
usage={...},
reasoning_content=getattr(message, "reasoning_content", None)
)
LLMResponse 数据结构
位置: nanobot/providers/base.py:8-28
@dataclass
class ToolCallRequest:
"""来自LLM的工具调用请求"""
id: str # 工具调用唯一标识
name: str # 工具名称
arguments: dict[str, Any] # 工具参数
@dataclass
class LLMResponse:
"""LLM提供商的响应"""
content: str | None
tool_calls: list[ToolCallRequest] = field(default_factory=list)
finish_reason: str = "stop"
usage: dict[str, int] = field(default_factory=dict)
reasoning_content: str | None = None # Kimi, DeepSeek-R1等推理内容
@property
def has_tool_calls(self) -> bool:
"""检查响应是否包含工具调用"""
return len(self.tool_calls) > 0
工具执行循环
Agent 主循环处理
位置: nanobot/agent/loop.py:202-231
# Agent循环
iteration = 0
final_content = None
while iteration < self.max_iterations:
iteration += 1
# 调用LLM
response = await self.provider.chat(
messages=messages,
tools=self.tools.get_definitions(),
model=self.model
)
# 处理工具调用
if response.has_tool_calls:
# 1. 添加assistant消息(包含tool_calls)到历史
tool_call_dicts = [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.name,
"arguments": json.dumps(tc.arguments)
}
}
for tc in response.tool_calls
]
messages = self.context.add_assistant_message(
messages, response.content, tool_call_dicts,
reasoning_content=response.reasoning_content,
)
# 2. 执行每个工具调用
for tool_call in response.tool_calls:
args_str = json.dumps(tool_call.arguments, ensure_ascii=False)
logger.info(f"Tool call: {tool_call.name}({args_str[:200]})")
# 执行工具
result = await self.tools.execute(
tool_call.name,
tool_call.arguments
)
# 3. 添加工具结果到消息历史
messages = self.context.add_tool_result(
messages, tool_call.id, tool_call.name, result
)
else:
# 无工具调用,LLM已完成任务
final_content = response.content
break
ToolRegistry 执行工具
位置: nanobot/agent/tools/registry.py:38-62
async def execute(self, name: str, params: dict[str, Any]) -> str:
"""
按名称执行工具及给定参数
"""
tool = self._tools.get(name)
if not tool:
return f"Error: Tool '{name}' not found"
try:
# 参数验证
errors = tool.validate_params(params)
if errors:
return f"Error: Invalid parameters for tool '{name}': " + "; ".join(errors)
# 执行工具
return await tool.execute(**params)
except Exception as e:
return f"Error executing {name}: {str(e)}"
参数验证
位置: nanobot/agent/tools/base.py:55-91
def validate_params(self, params: dict[str, Any]) -> list[str]:
"""验证工具参数是否符合JSON Schema"""
schema = self.parameters or {}
if schema.get("type", "object") != "object":
raise ValueError(f"Schema must be object type, got {schema.get('type')}")
return self._validate(params, {**schema, "type": "object"}, "")
def _validate(self, val: Any, schema: dict[str, Any], path: str) -> list[str]:
"""递归验证参数"""
t, label = schema.get("type"), path or "parameter"
# 类型检查
if t in self._TYPE_MAP and not isinstance(val, self._TYPE_MAP[t]):
return [f"{label} should be {t}"]
errors = []
# 枚举值检查
if "enum" in schema and val not in schema["enum"]:
errors.append(f"{label} must be one of {schema['enum']}")
# 数值范围检查
if t in ("integer", "number"):
if "minimum" in schema and val < schema["minimum"]:
errors.append(f"{label} must be >= {schema['minimum']}")
if "maximum" in schema and val > schema["maximum"]:
errors.append(f"{label} must be <= {schema['maximum']}")
# 字符串长度检查
if t == "string":
if "minLength" in schema and len(val) < schema["minLength"]:
errors.append(f"{label} must be at least {schema['minLength']} chars")
if "maxLength" in schema and len(val) > schema["maxLength"]:
errors.append(f"{label} must be at most {schema['maxLength']} chars")
# 对象属性检查
if t == "object":
props = schema.get("properties", {})
# 必填字段检查
for k in schema.get("required", []):
if k not in val:
errors.append(f"missing required {path + '.' + k if path else k}")
# 递归验证属性
for k, v in val.items():
if k in props:
errors.extend(self._validate(v, props[k], path + '.' + k if path else k))
# 数组元素检查
if t == "array" and "items" in schema:
for i, item in enumerate(val):
errors.extend(self._validate(item, schema["items"], f"{path}[{i}]" if path else f"[{i}]"))
return errors
消息格式规范
OpenAI Function Calling 标准格式
1. 用户消息
{
"role": "user",
"content": "帮我列出当前目录的文件"
}
2. Assistant消息(含tool_calls)
{
"role": "assistant",
"content": "我来帮你列出当前目录的文件...",
"tool_calls": [
{
"id": "call_abc123",
"type": "function",
"function": {
"name": "list_dir",
"arguments": "{\"path\": \".\"}"
}
}
]
}
3. Tool结果消息
{
"role": "tool",
"tool_call_id": "call_abc123",
"name": "list_dir",
"content": "📁 .\n📄 README.md\n📄 main.py\n📁 src/"
}
4. Assistant最终回复
{
"role": "assistant",
"content": "当前目录包含以下文件:\n- README.md\n- main.py\n- src/ (目录)"
}
ContextBuilder 消息管理
位置: nanobot/agent/context.py:179-235
def add_assistant_message(
self,
messages: list[dict[str, Any]],
content: str | None,
tool_calls: list[dict[str, Any]] | None = None,
reasoning_content: str | None = None,
) -> list[dict[str, Any]]:
"""添加assistant消息到消息列表"""
msg: dict[str, Any] = {
"role": "assistant",
"content": content or ""
}
if tool_calls:
msg["tool_calls"] = tool_calls
# 思维模型需要此字段
if reasoning_content:
msg["reasoning_content"] = reasoning_content
messages.append(msg)
return messages
def add_tool_result(
self,
messages: list[dict[str, Any]],
tool_call_id: str,
tool_name: str,
result: str
) -> list[dict[str, Any]]:
"""添加工具结果到消息列表"""
messages.append({
"role": "tool",
"tool_call_id": tool_call_id,
"name": tool_name,
"content": result
})
return messages
完整工作流程
用户请求到智能体响应的完整链路
完整代码流程示例
# 1. 用户发送消息
msg = InboundMessage(
channel="cli",
sender_id="user",
chat_id="direct",
content="帮我执行 ls -la 命令"
)
# 2. 构建上下文
messages = context.build_messages(
history=[],
current_message="帮我执行 ls -la 命令",
skill_names=None,
media=None,
channel="cli",
chat_id="direct"
)
# messages = [
# {"role": "system", "content": "包含技能摘要的系统提示词..."},
# {"role": "user", "content": "帮我执行 ls -la 命令"}
# ]
# 3. 调用LLM
response = await provider.chat(
messages=messages,
tools=[
{
"type": "function",
"function": {
"name": "exec",
"description": "Execute a shell command...",
"parameters": {
"type": "object",
"properties": {
"command": {"type": "string", "description": "The shell command to execute"}
},
"required": ["command"]
}
}
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "Read the contents of a file...",
"parameters": {...}
}
},
# ... 其他工具
],
model="gpt-4"
)
# 4. LLM响应(包含tool_calls)
# response.tool_calls = [
# ToolCallRequest(
# id="call_123",
# name="exec",
# arguments={"command": "ls -la"}
# )
# ]
# 5. 添加assistant消息到历史
messages = context.add_assistant_message(
messages,
content="我来帮你执行 ls -la 命令...",
tool_calls=[
{
"id": "call_123",
"type": "function",
"function": {
"name": "exec",
"arguments": '{"command": "ls -la"}'
}
}
]
)
# 6. 执行工具
result = await tools.execute("exec", {"command": "ls -la"})
# result = "total 32\ndrwxr-xr-x 5 user group 160 Feb 9 10:30 .\ndrwxr-xr-x 3 user group 96 Feb 9 10:25 src\n-rw-r--r-- 1 user group 1024 Feb 9 10:30 README.md\n..."
# 7. 添加工具结果到历史
messages = context.add_tool_result(
messages,
tool_call_id="call_123",
tool_name="exec",
result=result
)
# 8. 再次调用LLM(基于工具结果)
response = await provider.chat(
messages=messages,
tools=tools.get_definitions(),
model="gpt-4"
)
# 9. LLM返回最终文本(无tool_calls)
# response.content = "命令执行成功。当前目录包含:\n- README.md\n- src/\n..."
# response.tool_calls = []
# 10. 保存会话并返回
session.add_message("user", "帮我执行 ls -la 命令")
session.add_message("assistant", "命令执行成功。当前目录包含...")
sessions.save(session)
return OutboundMessage(
channel="cli",
chat_id="direct",
content="命令执行成功。当前目录包含:\n- README.md\n- src/..."
)
技能与工具的协同工作
核心文件索引
技能系统文件
| 文件路径 | 说明 | 关键类/函数 |
|---|---|---|
nanobot/agent/skills.py |
技能加载器 | SkillsLoader |
nanobot/agent/context.py |
上下文构建器 | ContextBuilder.build_system_prompt() |
nanobot/skills/ |
内置技能目录 | github, weather, tmux等 |
workspace/skills/ |
用户自定义技能 | 用户创建的技能 |
工具系统文件
| 文件路径 | 说明 | 关键类/函数 |
|---|---|---|
nanobot/agent/tools/base.py |
Tool抽象基类 | Tool |
nanobot/agent/tools/registry.py |
工具注册表 | ToolRegistry |
nanobot/agent/tools/shell.py |
Shell执行工具 | ExecTool |
nanobot/agent/tools/filesystem.py |
文件系统工具 | ReadFileTool, WriteFileTool, EditFileTool, ListDirTool |
nanobot/agent/tools/web.py |
Web工具 | WebSearchTool, WebFetchTool |
nanobot/agent/tools/message.py |
消息发送工具 | MessageTool |
nanobot/agent/tools/spawn.py |
子智能体工具 | SpawnTool |
nanobot/agent/tools/cron.py |
定时任务工具 | CronTool |
LLM集成文件
| 文件路径 | 说明 | 关键类/函数 |
|---|---|---|
nanobot/providers/base.py |
LLM提供商基类 | LLMProvider, LLMResponse, ToolCallRequest |
nanobot/providers/litellm_provider.py |
LiteLLM实现 | LiteLLMProvider.chat(), _parse_response() |
nanobot/agent/loop.py |
智能体循环 | AgentLoop._process_message() |
消息处理文件
| 文件路径 | 说明 | 关键类/函数 |
|---|---|---|
nanobot/agent/context.py |
上下文和消息管理 | ContextBuilder, add_assistant_message(), add_tool_result() |
nanobot/session/manager.py |
会话管理 | SessionManager |
nanobot/bus/events.py |
消息事件定义 | InboundMessage, OutboundMessage |
总结
技能(Skills)机制核心要点
- 模块化知识包: 技能是包含元数据和使用指令的Markdown文件,指导智能体如何完成特定任务
- 两级发现: workspace技能优先于内置技能,支持用户自定义
- 依赖管理: 通过元数据声明依赖(CLI工具、环境变量),加载时自动检查可用性
- 渐进式加载: 三级加载机制(元数据→SKILL.md→捆绑资源),高效管理上下文窗口
- XML摘要: 所有技能生成XML格式摘要,包含名称、描述、位置、可用性状态
工具调用(Tool Call)机制核心要点
- 标准格式: 遵循OpenAI Function Calling标准,所有工具转换为JSON Schema定义
- 参数验证: 基于JSON Schema进行严格的参数类型和约束检查
- 异步执行: 所有工具支持异步执行,避免阻塞
- 循环推理: LLM基于工具结果持续迭代,直到生成最终文本回复
- 消息历史: 完整的tool_calls和tool_results保存到消息历史,保持对话连贯性
两者协同工作
- 技能指导工具使用: 技能提供领域知识,指导智能体何时、如何使用特定工具
- 工具提供执行能力: 工具实现实际的外部交互(文件操作、命令执行等)
- LLM作为决策中心: LLM根据技能描述决定使用哪个工具,并生成正确的工具参数
更多推荐

所有评论(0)