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

完整工作流程

用户请求到智能体响应的完整链路

有tool_calls

无tool_calls

用户消息

AgentLoop._process_message

ContextBuilder.build_messages

构建系统提示词

添加身份信息

加载Bootstrap文件

加载Memory上下文

加载技能摘要

添加对话历史

添加当前消息

调用LLM provider.chat

LLM响应判断

解析ToolCallRequest

直接返回文本内容

添加assistant消息到历史

逐个执行工具调用

ToolRegistry.execute

参数验证

执行tool.execute

获取工具结果

添加tool结果到历史

达到最大迭代次数?

保存到session

返回OutboundMessage

完整代码流程示例

# 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/..."
)

技能与工具的协同工作

用户请求

需要特定技能?

查看技能摘要

直接使用工具

技能可用?

调用read_file读取SKILL.md

提示需要安装依赖

获得技能完整内容

根据技能指令使用工具

ToolRegistry.execute

工具执行结果

LLM基于结果继续推理


核心文件索引

技能系统文件

文件路径 说明 关键类/函数
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)机制核心要点

  1. 模块化知识包: 技能是包含元数据和使用指令的Markdown文件,指导智能体如何完成特定任务
  2. 两级发现: workspace技能优先于内置技能,支持用户自定义
  3. 依赖管理: 通过元数据声明依赖(CLI工具、环境变量),加载时自动检查可用性
  4. 渐进式加载: 三级加载机制(元数据→SKILL.md→捆绑资源),高效管理上下文窗口
  5. XML摘要: 所有技能生成XML格式摘要,包含名称、描述、位置、可用性状态

工具调用(Tool Call)机制核心要点

  1. 标准格式: 遵循OpenAI Function Calling标准,所有工具转换为JSON Schema定义
  2. 参数验证: 基于JSON Schema进行严格的参数类型和约束检查
  3. 异步执行: 所有工具支持异步执行,避免阻塞
  4. 循环推理: LLM基于工具结果持续迭代,直到生成最终文本回复
  5. 消息历史: 完整的tool_calls和tool_results保存到消息历史,保持对话连贯性

两者协同工作

  • 技能指导工具使用: 技能提供领域知识,指导智能体何时、如何使用特定工具
  • 工具提供执行能力: 工具实现实际的外部交互(文件操作、命令执行等)
  • LLM作为决策中心: LLM根据技能描述决定使用哪个工具,并生成正确的工具参数
Logo

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地,聚焦技能开发、插件实践与部署教程,为开发者提供可直接落地的方案、工具与交流平台,助力高效构建与落地 AI 应用

更多推荐