OpenClaw博客系列第4篇:ToolCalling工具调用机制

系列导读

欢迎来到OpenClaw博客系列的第四篇文章。在前三篇文章中,我们深入探讨了OpenClaw的核心架构、安全沙箱机制以及消息渠道接入方案。本篇文章将聚焦于OpenClaw最具创新性的特性之一——Tool Calling工具调用机制

Tool Calling(工具调用)是现代AI Agent系统的核心能力,它使大语言模型能够突破纯文本交互的限制,真正与外部世界进行交互。通过工具调用,AI Agent可以执行代码、查询数据库、调用API、操作文件系统,甚至控制物理设备。这种能力的引入,标志着AI从"对话助手"向"智能代理"的质的飞跃。

OpenClaw的Tool Calling机制在设计上追求三个核心目标:

第一,标准化。OpenClaw采用业界主流的工具定义规范,确保与各类LLM提供商的兼容性,同时为开发者提供一致的工具开发体验。

第二,安全性。工具调用涉及系统资源的访问,OpenClaw通过多层安全机制确保工具调用的可控性和可审计性,防止恶意或不当操作。

第三,可扩展性。OpenClaw提供了灵活的工具注册和发现机制,开发者可以轻松扩展新的工具能力,构建适合特定业务场景的工具生态。

本文将从基础概念出发,逐步深入到实现细节,涵盖工具定义规范、执行流程、安全机制、高级特性以及实战案例。无论你是OpenClaw的使用者还是贡献者,都能从本文中获得对Tool Calling机制的全面理解。


第一章:Tool Calling基础概念

1.1 什么是Tool Calling

Tool Calling,中文常译为"工具调用"或"函数调用",是一种让大语言模型(LLM)能够调用外部工具或函数的机制。在传统的LLM交互中,模型只能生成文本响应,无法直接执行任何实际操作。Tool Calling的引入打破了这一限制,使LLM能够:

  1. 获取实时信息:查询当前天气、股票价格、新闻资讯等动态数据
  2. 执行计算任务:进行复杂的数学运算、数据分析、代码执行
  3. 操作外部系统:调用API、操作数据库、发送邮件、创建文件
  4. 控制物理设备:在IoT场景中控制智能家居、工业设备等

从技术实现的角度,Tool Calling的工作流程可以概括为以下几个步骤:

步骤一:工具定义
开发者预先定义一组工具,包括工具名称、功能描述、参数规范等。这些定义通常采用JSON Schema格式,确保LLM能够理解工具的用途和使用方法。

步骤二:工具注入
在向LLM发送请求时,将工具定义作为上下文信息注入到提示词中。LLM会根据这些信息判断是否需要调用工具来完成任务。

步骤三:决策生成
LLM分析用户的请求,结合工具定义,决定是否需要调用工具。如果需要,LLM会生成一个结构化的工具调用请求,包含工具名称和参数值。

步骤四:工具执行
系统接收到LLM的工具调用请求后,执行相应的工具函数,并将执行结果返回给LLM。

步骤五:结果整合
LLM将工具执行结果整合到最终的响应中,生成对用户有意义的回答。

让我们通过一个简单的例子来理解这个过程:

用户:北京现在的天气怎么样?

LLM分析:用户询问实时天气信息,我无法直接获取,但可以使用weather_get工具。

LLM生成工具调用:
{
  "tool": "weather_get",
  "parameters": {
    "city": "北京"
  }
}

系统执行:调用天气API,获取北京当前天气数据

工具返回:
{
  "temperature": "25°C",
  "condition": "晴",
  "humidity": "45%",
  "wind": "东南风3级"
}

LLM整合:北京现在的天气是晴天,气温25°C,湿度45%,东南风3级。总体来说是个不错的天气。

从这个例子可以看出,Tool Calling使LLM从"只能说话"进化为"能够做事",这是AI Agent能力的关键突破。

1.2 Tool Calling的发展历程

Tool Calling并非一蹴而就的技术,它经历了从概念提出到标准化实现的演进过程。了解这一历程有助于我们更好地理解当前的技术形态和未来的发展方向。

早期探索阶段(2020-2022)

在GPT-3发布初期,研究者们就开始探索如何让LLM与外部工具交互。早期的尝试主要包括:

Prompt Engineering方法
通过精心设计的提示词,引导LLM生成特定格式的输出,然后通过后处理解析这些输出并执行相应操作。例如:

如果你需要查询天气,请输出格式:[QUERY_WEATHER:城市名]
如果你需要计算,请输出格式:[CALCULATE:表达式]

这种方法虽然简单,但存在明显缺陷:格式解析容易出错,复杂任务难以处理,缺乏标准化。

LangChain的早期工具集成
LangChain框架在2022年底开始提供工具集成能力,通过定义"Tool"抽象,让LLM能够调用预定义的函数。这为后来的标准化奠定了基础。

标准化阶段(2023)

2023年是Tool Calling技术标准化的重要一年,多个LLM提供商推出了官方的Function Calling支持:

OpenAI Function Calling(2023年6月)
OpenAI在GPT-3.5和GPT-4中引入了原生的Function Calling能力。开发者可以通过API传递函数定义,模型会自动决定是否调用函数,并生成符合规范的JSON参数。这标志着Tool Calling从"黑科技"走向"标准功能"。

Anthropic Tool Use
Claude模型提供了类似的工具使用能力,强调安全性和可控性,支持更复杂的工具使用场景。

Google Gemini Function Calling
Google的Gemini模型也支持函数调用,与OpenAI的规范基本兼容,但在某些细节上有所不同。

生态成熟阶段(2024至今)

随着各大LLM提供商的支持,Tool Calling的生态系统快速成熟:

标准化协议
业界开始形成统一的工具定义规范,虽然各家实现略有差异,但核心概念趋于一致。JSON Schema成为工具参数定义的事实标准。

工具生态
大量预构建的工具库涌现,涵盖API调用、数据库操作、文件处理、代码执行等常见场景。开发者可以像使用npm包一样使用这些工具。

框架支持
主流的AI Agent框架(如LangChain、AutoGPT、OpenClaw)都提供了完善的Tool Calling支持,降低了开发门槛。

安全增强
工具调用的安全机制日益完善,包括权限控制、执行沙箱、审计日志等,使Tool Calling能够在企业环境中安全使用。

1.3 OpenClaw的Tool Calling实现

OpenClaw在吸收业界最佳实践的基础上,构建了一套完整的Tool Calling实现方案。其核心设计理念可以概括为"安全、灵活、可控"。

架构概览

OpenClaw的Tool Calling机制由以下几个核心组件构成:

工具注册中心(Tool Registry)
负责管理所有可用工具的定义和元数据。工具开发者通过注册中心发布工具,系统通过注册中心发现和加载工具。

工具调度器(Tool Dispatcher)
接收来自LLM的工具调用请求,根据请求内容选择合适的工具执行器,并协调执行过程。

工具执行器(Tool Executor)
实际执行工具调用的组件。OpenClaw支持多种执行模式,包括本地执行、沙箱执行、远程执行等。

安全网关(Security Gateway)
在工具执行前后进行安全检查,包括权限验证、参数校验、资源限制等,确保工具调用的安全性。

结果处理器(Result Processor)
处理工具执行的返回结果,进行格式转换、数据脱敏、错误处理等,将结果转换为LLM可理解的格式。

核心特性

OpenClaw的Tool Calling实现具有以下核心特性:

多模型兼容
OpenClaw支持多种LLM提供商的Tool Calling格式,包括OpenAI、Anthropic、Google等。系统会自动适配不同模型的工具调用协议,开发者无需关心底层差异。

动态工具发现
工具可以在运行时动态注册和卸载,无需重启系统。这支持了灵活的工具管理和热更新能力。

分层权限控制
工具调用遵循最小权限原则,每个工具都有独立的权限配置。系统支持基于角色、基于用户、基于会话的多维度权限控制。

执行隔离
工具在独立的执行环境中运行,与主系统隔离。即使工具执行出现问题,也不会影响系统的整体稳定性。

完整审计
所有工具调用都有详细的审计日志,包括调用时间、调用者、参数、结果、执行时长等信息,支持事后追溯和分析。

异步执行
对于耗时较长的工具调用,OpenClaw支持异步执行模式,避免阻塞主流程。异步任务的状态和结果可以通过回调或轮询获取。

与其他模块的集成

Tool Calling机制与OpenClaw的其他核心模块紧密集成:

与消息渠道的集成
工具调用结果可以通过各种消息渠道返回给用户,支持文本、图片、文件等多种消息类型。

与安全沙箱的集成
工具执行在安全沙箱中进行,受沙箱的资源限制和安全策略约束。

与对话管理的集成
工具调用是对话流程的一部分,对话管理器负责协调工具调用与对话状态的同步。

与知识库的集成
工具可以访问知识库,获取业务知识辅助执行。同时,工具执行结果可以沉淀到知识库中,形成知识闭环。


第二章:工具定义规范

工具定义是Tool Calling机制的基础。一个清晰、规范的工具定义能够帮助LLM准确理解工具的功能和使用方法,从而做出正确的调用决策。OpenClaw采用JSON Schema作为工具定义的核心规范,并在此基础上进行了扩展,以满足更复杂的应用场景。

2.1 JSON Schema定义

JSON Schema是一种用于描述JSON数据结构的规范,它提供了一套完整的语法来定义数据的类型、格式、约束条件等。在Tool Calling中,JSON Schema主要用于两个方面:定义工具的元信息(如名称、描述)和定义工具的参数规范。

基本结构

一个完整的工具定义包含以下核心字段:

{
  "name": "tool_name",
  "description": "工具的功能描述,LLM会根据这个描述判断是否需要调用此工具",
  "parameters": {
    "type": "object",
    "properties": {
      "param1": {
        "type": "string",
        "description": "参数1的描述"
      },
      "param2": {
        "type": "number",
        "description": "参数2的描述"
      }
    },
    "required": ["param1"]
  }
}

让我们详细解释每个字段的作用:

name(工具名称)
工具的唯一标识符,用于在调用时指定要执行的工具。命名应遵循以下原则:

  • 使用小写字母和下划线,如get_weathersend_email
  • 具有描述性,能够从名称推断出工具的功能
  • 避免使用保留字和特殊字符
  • 建议采用动词_名词的命名模式

description(工具描述)
对工具功能的详细说明,这是LLM判断是否调用工具的关键依据。好的描述应该:

  • 清晰说明工具的功能和用途
  • 说明工具的适用场景和限制条件
  • 提供使用示例(可选但推荐)
  • 避免模糊和歧义的表述

例如,一个天气查询工具的描述可以这样写:

{
  "description": "查询指定城市的当前天气信息,包括温度、湿度、风速、天气状况等。适用于用户询问天气、出行建议等场景。注意:仅支持中国主要城市,国外城市可能无法查询。"
}

parameters(参数定义)
使用JSON Schema定义工具的输入参数。参数定义决定了LLM生成调用请求时需要提供哪些信息。

参数定义详解

参数定义遵循JSON Schema规范,支持丰富的类型和约束:

基本类型

{
  "properties": {
    "text_param": {
      "type": "string",
      "description": "字符串类型参数"
    },
    "number_param": {
      "type": "number",
      "description": "数字类型参数(整数或浮点数)"
    },
    "integer_param": {
      "type": "integer",
      "description": "整数类型参数"
    },
    "boolean_param": {
      "type": "boolean",
      "description": "布尔类型参数"
    }
  }
}

枚举类型

当参数只能取特定的几个值时,可以使用枚举:

{
  "properties": {
    "unit": {
      "type": "string",
      "enum": ["celsius", "fahrenheit"],
      "description": "温度单位"
    },
    "sort_order": {
      "type": "string",
      "enum": ["asc", "desc"],
      "description": "排序方式"
    }
  }
}

数组类型

{
  "properties": {
    "tags": {
      "type": "array",
      "items": {
        "type": "string"
      },
      "description": "标签列表"
    },
    "ids": {
      "type": "array",
      "items": {
        "type": "integer"
      },
      "minItems": 1,
      "maxItems": 100,
      "description": "ID列表,最少1个,最多100个"
    }
  }
}

嵌套对象

参数可以是复杂的嵌套结构:

{
  "properties": {
    "filter": {
      "type": "object",
      "properties": {
        "date_range": {
          "type": "object",
          "properties": {
            "start": {"type": "string", "format": "date"},
            "end": {"type": "string", "format": "date"}
          }
        },
        "categories": {
          "type": "array",
          "items": {"type": "string"}
        }
      },
      "description": "查询过滤条件"
    }
  }
}

约束条件

JSON Schema支持多种约束条件,确保参数的有效性:

{
  "properties": {
    "username": {
      "type": "string",
      "minLength": 3,
      "maxLength": 20,
      "pattern": "^[a-zA-Z0-9_]+$",
      "description": "用户名,3-20个字符,只能包含字母、数字和下划线"
    },
    "age": {
      "type": "integer",
      "minimum": 0,
      "maximum": 150,
      "description": "年龄,0-150之间"
    },
    "email": {
      "type": "string",
      "format": "email",
      "description": "邮箱地址"
    }
  }
}

必填参数

通过required字段指定哪些参数是必须提供的:

{
  "type": "object",
  "properties": {
    "query": {
      "type": "string",
      "description": "搜索关键词"
    },
    "limit": {
      "type": "integer",
      "default": 10,
      "description": "返回结果数量,默认10"
    }
  },
  "required": ["query"]
}

在上面的例子中,query是必填参数,而limit是可选参数,有默认值10。

2.2 参数类型系统

OpenClaw在标准JSON Schema的基础上,扩展了一套参数类型系统,提供更丰富的类型支持和更严格的类型检查。

扩展类型

日期时间类型

{
  "properties": {
    "created_at": {
      "type": "string",
      "format": "date-time",
      "description": "创建时间,ISO 8601格式"
    },
    "birthday": {
      "type": "string",
      "format": "date",
      "description": "生日,YYYY-MM-DD格式"
    },
    "meeting_time": {
      "type": "string",
      "format": "time",
      "description": "会议时间,HH:mm:ss格式"
    }
  }
}

网络相关类型

{
  "properties": {
    "website": {
      "type": "string",
      "format": "uri",
      "description": "网站URL"
    },
    "api_endpoint": {
      "type": "string",
      "format": "uri-reference",
      "description": "API端点,可以是相对路径"
    },
    "ip_address": {
      "type": "string",
      "format": "ipv4",
      "description": "IPv4地址"
    }
  }
}

文件路径类型

OpenClaw定义了特殊的文件路径类型,支持路径验证和安全检查:

{
  "properties": {
    "file_path": {
      "type": "string",
      "format": "file-path",
      "description": "文件路径,系统会验证路径是否存在且可访问"
    },
    "output_dir": {
      "type": "string",
      "format": "directory-path",
      "description": "输出目录路径"
    }
  }
}

自定义类型

开发者可以定义自定义类型,并注册类型验证器:

{
  "properties": {
    "phone": {
      "type": "string",
      "format": "phone-number",
      "description": "手机号码"
    },
    "id_card": {
      "type": "string",
      "format": "id-card-cn",
      "description": "中国身份证号码"
    }
  }
}
类型转换

OpenClaw支持自动类型转换,当LLM生成的参数类型与定义不完全匹配时,系统会尝试进行合理的转换:

字符串转数字

输入: "123" -> 定义类型: integer -> 转换结果: 123
输入: "3.14" -> 定义类型: number -> 转换结果: 3.14

数字转字符串

输入: 123 -> 定义类型: string -> 转换结果: "123"

字符串转布尔

输入: "true" -> 定义类型: boolean -> 转换结果: true
输入: "false" -> 定义类型: boolean -> 转换结果: false
输入: "yes" -> 定义类型: boolean -> 转换结果: true
输入: "no" -> 定义类型: boolean -> 转换结果: false

数组转换

输入: "a,b,c" -> 定义类型: array -> 转换结果: ["a", "b", "c"]
输入: "item1;item2" -> 定义类型: array -> 转换结果: ["item1", "item2"]

类型转换遵循"尽力而为"的原则,如果转换失败,系统会返回验证错误,要求LLM重新生成参数。

类型验证流程

当LLM生成工具调用请求后,OpenClaw会进行严格的类型验证:

第一步:结构验证
检查参数是否包含所有必填字段,是否有多余字段(根据配置决定是否允许)。

第二步:类型验证
检查每个参数值的类型是否符合定义,尝试自动类型转换。

第三步:约束验证
检查参数值是否满足定义的约束条件(如最小值、最大值、正则表达式等)。

第四步:语义验证
对于特定格式的参数(如日期、邮箱、文件路径等),进行语义级别的验证。

验证失败时,系统会生成详细的错误信息,反馈给LLM进行修正:

{
  "error": "parameter_validation_failed",
  "details": [
    {
      "path": "/email",
      "message": "Invalid email format: 'test@' is not a valid email address"
    },
    {
      "path": "/age",
      "message": "Value 200 exceeds maximum allowed value 150"
    }
  ]
}

2.3 工具注册机制

工具注册是将工具定义和实现关联起来的过程。OpenClaw提供了灵活的注册机制,支持多种注册方式和生命周期管理。

注册方式

静态注册

在系统启动时通过配置文件注册工具:

# openclaw-config.yaml
tools:
  - name: weather_get
    module: openclaw.tools.weather
    class: WeatherTool
    enabled: true

  - name: email_send
    module: openclaw.tools.email
    class: EmailTool
    enabled: true
    config:
      smtp_host: "smtp.example.com"
      smtp_port: 587

静态注册适合核心工具,这些工具在系统运行期间始终可用。

动态注册

在运行时通过API注册工具:

from openclaw import ToolRegistry

# 创建工具定义
tool_def = {
    "name": "custom_calculator",
    "description": "执行自定义数学计算",
    "parameters": {
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "数学表达式,如 '2+3*4'"
            }
        },
        "required": ["expression"]
    }
}

# 注册工具
registry = ToolRegistry.get_instance()
registry.register(
    definition=tool_def,
    handler=calculator_handler,
    permissions=["calc:execute"]
)

动态注册适合插件式工具,可以根据需要加载和卸载。

装饰器注册

使用装饰器简化工具注册:

from openclaw import tool

@tool(
    name="search_web",
    description="在网络上搜索信息",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "搜索关键词"},
            "num_results": {"type": "integer", "default": 5}
        },
        "required": ["query"]
    }
)
def search_web(query: str, num_results: int = 5):
    """执行网络搜索"""
    # 实现搜索逻辑
    results = search_engine.search(query, max_results=num_results)
    return {"results": results}

装饰器方式最简洁,适合快速开发。

工具元数据

除了基本的定义信息,工具还可以携带丰富的元数据:

tool_metadata = {
    "name": "database_query",
    "version": "2.1.0",
    "author": "OpenClaw Team",
    "category": "database",
    "tags": ["sql", "query", "database"],
    "dependencies": ["pymysql>=1.0.0"],
    "permissions": ["db:read", "db:write"],
    "rate_limit": {
        "requests_per_minute": 60,
        "tokens_per_request": 1000
    },
    "timeout": 30,  # 秒
    "retry_policy": {
        "max_retries": 3,
        "backoff": "exponential"
    },
    "examples": [
        {
            "description": "查询用户表",
            "parameters": {"table": "users", "limit": 10},
            "result": {"rows": [...], "count": 10}
        }
    ]
}

这些元数据用于:

  • 工具发现和分类
  • 权限和限流配置
  • 错误处理和重试策略
  • 文档生成和示例展示
工具生命周期

注册后的工具经历以下生命周期状态:

Registered(已注册)
工具定义已加载到注册中心,但尚未初始化。此时工具不可用。

Initialized(已初始化)
工具完成初始化(如建立数据库连接、加载配置等),可以接受调用请求。

Active(活跃)
工具正在被使用,系统会跟踪调用次数、成功率等指标。

Deprecated(已弃用)
工具标记为弃用,仍然可用但不推荐使用。系统会在调用时发出警告。

Disabled(已禁用)
工具被禁用,无法调用。可能是因为配置错误、依赖缺失或管理员手动禁用。

Unregistered(已注销)
工具从注册中心移除,所有相关资源已释放。

生命周期管理API:

# 初始化工具
registry.initialize_tool("database_query")

# 禁用工具
registry.disable_tool("database_query", reason="维护中")

# 启用工具
registry.enable_tool("database_query")

# 弃用工具
registry.deprecate_tool("old_tool", successor="new_tool")

# 注销工具
registry.unregister("old_tool")

2.4 工具发现与加载

工具发现是系统自动识别和加载可用工具的过程。OpenClaw支持多种发现机制,满足不同的部署场景。

目录扫描发现

系统扫描指定目录,自动发现和加载工具模块:

# 配置文件
tool_discovery:
  scan_directories:
    - "/opt/openclaw/tools"
    - "/var/openclaw/plugins"
  auto_load: true
  patterns:
    - "tool_*.py"
    - "*_tool.py"

扫描到的工具模块需要遵循特定的结构:

# tools/weather_tool.py
from openclaw import BaseTool

class WeatherTool(BaseTool):
    """天气查询工具"""
    
    name = "weather_get"
    description = "查询指定城市的天气信息"
    
    def get_definition(self):
        return {
            "name": self.name,
            "description": self.description,
            "parameters": {...}
        }
    
    def execute(self, **params):
        # 实现逻辑
        pass

# 工具导出
__all__ = ["WeatherTool"]
包管理器发现

通过包管理器(如pip)安装的工具,OpenClaw可以自动发现:

# setup.py (工具包)
from setuptools import setup

setup(
    name="openclaw-tool-slack",
    version="1.0.0",
    entry_points={
        "openclaw.tools": [
            "slack = openclaw_tool_slack:SlackTool"
        ]
    }
)

安装后,OpenClaw会自动扫描openclaw.tools入口点,加载对应的工具。

远程发现

对于分布式部署,工具可以从远程服务发现:

tool_discovery:
  remote_registry:
    url: "https://tools.openclaw.io/api/v1/registry"
    auth_token: "${TOOL_REGISTRY_TOKEN}"
    sync_interval: 300  # 每5分钟同步一次

远程发现支持:

  • 工具版本管理
  • 工具权限同步
  • 工具使用统计
  • 工具健康检查
工具加载策略

发现工具后,系统需要决定何时加载。OpenClaw支持多种加载策略:

预加载(Eager Loading)
系统启动时加载所有已发现的工具。适合工具数量较少、内存充足的场景。

tool_loading:
  strategy: "eager"
  preload_all: true

延迟加载(Lazy Loading)
工具在首次被调用时才加载。适合工具数量多、内存有限的场景。

tool_loading:
  strategy: "lazy"
  cache_loaded: true
  max_cache_size: 100

按需加载(On-Demand Loading)
根据会话或用户需求动态加载工具。适合多租户场景,不同用户可用的工具集不同。

tool_loading:
  strategy: "on-demand"
  user_tool_profiles:
    - profile: "basic"
      tools: ["weather_get", "search_web", "calculator"]
    - profile: "advanced"
      tools: ["database_query", "code_execute", "file_operations"]
工具加载流程

完整的工具加载流程如下:

1. 发现阶段
   ├── 扫描配置目录
   ├── 检查包入口点
   └── 查询远程注册中心

2. 验证阶段
   ├── 检查工具定义完整性
   ├── 验证依赖是否满足
   └── 检查权限配置

3. 加载阶段
   ├── 导入工具模块
   ├── 实例化工具类
   └── 注册到工具注册中心

4. 初始化阶段
   ├── 调用工具初始化方法
   ├── 建立外部连接
   └── 加载工具配置

5. 就绪阶段
   └── 工具可接受调用请求

第三章:工具执行流程

工具执行是Tool Calling机制的核心环节。当LLM决定调用工具后,系统需要安全、高效地执行工具调用,并将结果返回给LLM。OpenClaw设计了一套完整的工具执行流程,涵盖从请求接收到结果返回的全过程。

3.1 LLM决策调用

LLM决策调用是工具执行的起点。当用户发送消息后,系统会将可用的工具定义注入到提示词中,LLM根据用户请求和工具描述决定是否调用工具。

工具定义注入

在向LLM发送请求时,系统会将工具定义转换为特定格式注入到请求中。不同LLM提供商的格式略有差异:

OpenAI格式

{
  "model": "gpt-4",
  "messages": [
    {"role": "user", "content": "北京现在的天气怎么样?"}
  ],
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "get_weather",
        "description": "查询指定城市的当前天气信息",
        "parameters": {
          "type": "object",
          "properties": {
            "city": {
              "type": "string",
              "description": "城市名称"
            },
            "unit": {
              "type": "string",
              "enum": ["celsius", "fahrenheit"],
              "description": "温度单位"
            }
          },
          "required": ["city"]
        }
      }
    }
  ],
  "tool_choice": "auto"
}

Anthropic格式

{
  "model": "claude-3-opus",
  "messages": [
    {"role": "user", "content": "北京现在的天气怎么样?"}
  ],
  "tools": [
    {
      "name": "get_weather",
      "description": "查询指定城市的当前天气信息",
      "input_schema": {
        "type": "object",
        "properties": {
          "city": {
            "type": "string",
            "description": "城市名称"
          }
        },
        "required": ["city"]
      }
    }
  ]
}

OpenClaw通过适配层自动处理不同格式的差异,开发者只需定义一次工具,即可在所有支持的LLM上使用。

工具选择策略

tool_choice参数控制LLM的工具选择行为:

auto(自动选择)
LLM自主决定是否调用工具,这是默认行为。

{"tool_choice": "auto"}

none(不使用工具)
强制LLM不使用任何工具,只生成文本响应。

{"tool_choice": "none"}

required(必须使用工具)
强制LLM至少调用一个工具。

{"tool_choice": "required"}

指定工具
强制LLM调用特定的工具。

{"tool_choice": {"type": "function", "function": {"name": "get_weather"}}}
LLM响应解析

当LLM决定调用工具时,会返回一个工具调用请求:

{
  "id": "call_abc123",
  "object": "chat.completion",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": null,
        "tool_calls": [
          {
            "id": "call_xyz789",
            "type": "function",
            "function": {
              "name": "get_weather",
              "arguments": "{\"city\": \"北京\", \"unit\": \"celsius\"}"
            }
          }
        ]
      },
      "finish_reason": "tool_calls"
    }
  ]
}

OpenClaw的响应解析器会提取关键信息:

class ToolCallParser:
    """工具调用响应解析器"""
    
    def parse(self, response: dict) -> List[ToolCallRequest]:
        """解析LLM响应,提取工具调用请求"""
        tool_calls = []
        
        for choice in response.get("choices", []):
            message = choice.get("message", {})
            
            if message.get("tool_calls"):
                for tc in message["tool_calls"]:
                    request = ToolCallRequest(
                        call_id=tc["id"],
                        tool_name=tc["function"]["name"],
                        arguments=json.loads(tc["function"]["arguments"]),
                        metadata={
                            "model": response.get("model"),
                            "finish_reason": choice.get("finish_reason")
                        }
                    )
                    tool_calls.append(request)
        
        return tool_calls
多工具调用处理

LLM可能在一个响应中请求调用多个工具:

{
  "tool_calls": [
    {
      "id": "call_1",
      "function": {"name": "get_weather", "arguments": "{\"city\": \"北京\"}"}
    },
    {
      "id": "call_2",
      "function": {"name": "get_weather", "arguments": "{\"city\": \"上海\"}"}
    }
  ]
}

OpenClaw支持并行执行多个独立的工具调用,提高执行效率:

async def execute_parallel(self, calls: List[ToolCallRequest]) -> List[ToolCallResult]:
    """并行执行多个工具调用"""
    tasks = [self.execute_single(call) for call in calls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    return [
        result if isinstance(result, ToolCallResult) 
        else ToolCallResult.error(call_id, str(result))
        for call_id, result in zip([c.call_id for c in calls], results)
    ]

3.2 参数校验与预处理

在执行工具之前,系统会对LLM生成的参数进行严格的校验和预处理,确保参数的有效性和安全性。

参数校验流程
class ParameterValidator:
    """参数校验器"""
    
    def validate(self, tool_def: ToolDefinition, arguments: dict) -> ValidationResult:
        """校验工具调用参数"""
        errors = []
        warnings = []
        sanitized = {}
        
        schema = tool_def.parameters
        
        # 1. 检查必填参数
        required = schema.get("required", [])
        for field in required:
            if field not in arguments:
                errors.append(ValidationError(
                    path=f"/{field}",
                    code="required_field_missing",
                    message=f"Required field '{field}' is missing"
                ))
        
        # 2. 检查未知字段
        properties = schema.get("properties", {})
        for key in arguments:
            if key not in properties:
                if schema.get("additionalProperties", True):
                    sanitized[key] = arguments[key]
                else:
                    warnings.append(ValidationWarning(
                        path=f"/{key}",
                        code="unknown_field",
                        message=f"Unknown field '{key}' will be ignored"
                    ))
        
        # 3. 类型校验和转换
        for key, value in arguments.items():
            if key in properties:
                prop_def = properties[key]
                try:
                    sanitized[key] = self._validate_type(key, value, prop_def)
                except ValidationError as e:
                    errors.append(e)
        
        # 4. 约束校验
        for key, value in sanitized.items():
            if key in properties:
                prop_def = properties[key]
                constraint_errors = self._validate_constraints(key, value, prop_def)
                errors.extend(constraint_errors)
        
        return ValidationResult(
            valid=len(errors) == 0,
            sanitized=sanitized,
            errors=errors,
            warnings=warnings
        )
    
    def _validate_type(self, key: str, value: Any, prop_def: dict) -> Any:
        """验证并转换参数类型"""
        expected_type = prop_def.get("type")
        
        # 字符串类型
        if expected_type == "string":
            if isinstance(value, str):
                return value
            return str(value)
        
        # 整数类型
        elif expected_type == "integer":
            if isinstance(value, int) and not isinstance(value, bool):
                return value
            if isinstance(value, (str, float)):
                try:
                    return int(float(value))
                except (ValueError, TypeError):
                    raise ValidationError(
                        path=f"/{key}",
                        code="type_mismatch",
                        message=f"Cannot convert '{value}' to integer"
                    )
            raise ValidationError(
                path=f"/{key}",
                code="type_mismatch",
                message=f"Expected integer, got {type(value).__name__}"
            )
        
        # 数字类型
        elif expected_type == "number":
            if isinstance(value, (int, float)) and not isinstance(value, bool):
                return float(value)
            if isinstance(value, str):
                try:
                    return float(value)
                except ValueError:
                    raise ValidationError(
                        path=f"/{key}",
                        code="type_mismatch",
                        message=f"Cannot convert '{value}' to number"
                    )
            raise ValidationError(
                path=f"/{key}",
                code="type_mismatch",
                message=f"Expected number, got {type(value).__name__}"
            )
        
        # 布尔类型
        elif expected_type == "boolean":
            if isinstance(value, bool):
                return value
            if isinstance(value, str):
                if value.lower() in ("true", "yes", "1"):
                    return True
                elif value.lower() in ("false", "no", "0"):
                    return False
            raise ValidationError(
                path=f"/{key}",
                code="type_mismatch",
                message=f"Cannot convert '{value}' to boolean"
            )
        
        # 数组类型
        elif expected_type == "array":
            if isinstance(value, list):
                return value
            if isinstance(value, str):
                # 尝试解析逗号分隔的列表
                return [item.strip() for item in value.split(",")]
            raise ValidationError(
                path=f"/{key}",
                code="type_mismatch",
                message=f"Expected array, got {type(value).__name__}"
            )
        
        # 对象类型
        elif expected_type == "object":
            if isinstance(value, dict):
                return value
            if isinstance(value, str):
                try:
                    return json.loads(value)
                except json.JSONDecodeError:
                    raise ValidationError(
                        path=f"/{key}",
                        code="type_mismatch",
                        message=f"Cannot parse '{value}' as JSON object"
                    )
            raise ValidationError(
                path=f"/{key}",
                code="type_mismatch",
                message=f"Expected object, got {type(value).__name__}"
            )
        
        return value
    
    def _validate_constraints(self, key: str, value: Any, prop_def: dict) -> List[ValidationError]:
        """验证参数约束"""
        errors = []
        
        # 字符串约束
        if isinstance(value, str):
            if "minLength" in prop_def and len(value) < prop_def["minLength"]:
                errors.append(ValidationError(
                    path=f"/{key}",
                    code="min_length_violation",
                    message=f"String length {len(value)} is less than minimum {prop_def['minLength']}"
                ))
            if "maxLength" in prop_def and len(value) > prop_def["maxLength"]:
                errors.append(ValidationError(
                    path=f"/{key}",
                    code="max_length_violation",
                    message=f"String length {len(value)} exceeds maximum {prop_def['maxLength']}"
                ))
            if "pattern" in prop_def:
                if not re.match(prop_def["pattern"], value):
                    errors.append(ValidationError(
                        path=f"/{key}",
                        code="pattern_mismatch",
                        message=f"Value '{value}' does not match pattern '{prop_def['pattern']}'"
                    ))
        
        # 数字约束
        if isinstance(value, (int, float)):
            if "minimum" in prop_def and value < prop_def["minimum"]:
                errors.append(ValidationError(
                    path=f"/{key}",
                    code="minimum_violation",
                    message=f"Value {value} is less than minimum {prop_def['minimum']}"
                ))
            if "maximum" in prop_def and value > prop_def["maximum"]:
                errors.append(ValidationError(
                    path=f"/{key}",
                    code="maximum_violation",
                    message=f"Value {value} exceeds maximum {prop_def['maximum']}"
                ))
        
        # 枚举约束
        if "enum" in prop_def and value not in prop_def["enum"]:
            errors.append(ValidationError(
                path=f"/{key}",
                code="enum_mismatch",
                message=f"Value '{value}' is not one of allowed values: {prop_def['enum']}"
            ))
        
        # 数组约束
        if isinstance(value, list):
            if "minItems" in prop_def and len(value) < prop_def["minItems"]:
                errors.append(ValidationError(
                    path=f"/{key}",
                    code="min_items_violation",
                    message=f"Array has {len(value)} items, minimum is {prop_def['minItems']}"
                ))
            if "maxItems" in prop_def and len(value) > prop_def["maxItems"]:
                errors.append(ValidationError(
                    path=f"/{key}",
                    code="max_items_violation",
                    message=f"Array has {len(value)} items, maximum is {prop_def['maxItems']}"
                ))
        
        return errors
参数预处理

参数预处理在验证通过后执行,用于规范化参数值和注入上下文信息:

class ParameterPreprocessor:
    """参数预处理器"""
    
    def preprocess(self, tool_def: ToolDefinition, arguments: dict, context: ExecutionContext) -> dict:
        """预处理参数"""
        processed = dict(arguments)
        
        # 1. 填充默认值
        properties = tool_def.parameters.get("properties", {})
        for key, prop_def in properties.items():
            if key not in processed and "default" in prop_def:
                processed[key] = prop_def["default"]
        
        # 2. 应用参数转换器
        for key, value in processed.items():
            if key in properties:
                transformer = self._get_transformer(properties[key])
                if transformer:
                    processed[key] = transformer(value, context)
        
        # 3. 注入上下文参数
        if tool_def.inject_context:
            for param_name, context_path in tool_def.context_mappings.items():
                processed[param_name] = self._resolve_context(context, context_path)
        
        # 4. 敏感信息处理
        for key in properties:
            if properties[key].get("sensitive", False):
                processed[f"_{key}_original"] = processed.get(key)
                processed[key] = self._mask_sensitive(processed.get(key))
        
        return processed
    
    def _get_transformer(self, prop_def: dict) -> Optional[Callable]:
        """获取参数转换器"""
        format_type = prop_def.get("format")
        
        transformers = {
            "date-time": self._transform_datetime,
            "date": self._transform_date,
            "uri": self._transform_uri,
            "email": self._transform_email,
            "file-path": self._transform_file_path
        }
        
        return transformers.get(format_type)
    
    def _transform_datetime(self, value: str, context: ExecutionContext) -> datetime:
        """转换日期时间格式"""
        if isinstance(value, datetime):
            return value
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    
    def _transform_file_path(self, value: str, context: ExecutionContext) -> str:
        """转换文件路径,确保在允许的目录范围内"""
        abs_path = os.path.abspath(value)
        allowed_dirs = context.allowed_directories
        
        if not any(abs_path.startswith(allowed_dir) for allowed_dir in allowed_dirs):
            raise SecurityError(f"File path '{value}' is outside allowed directories")
        
        return abs_path

3.3 Sandbox隔离执行

工具执行的安全性是OpenClaw设计的核心考量。所有工具调用都在隔离的沙箱环境中执行,确保工具执行不会影响系统稳定性或造成安全风险。

沙箱架构
┌─────────────────────────────────────────────────────────────┐
│                     OpenClaw主进程                           │
├─────────────────────────────────────────────────────────────┤
│  ┌─────────────────┐  ┌─────────────────┐                  │
│  │  Tool Dispatcher │  │ Security Gateway │                 │
│  └────────┬────────┘  └────────┬────────┘                  │
│           │                    │                            │
│           ▼                    ▼                            │
│  ┌─────────────────────────────────────────────────────────┐│
│  │                   Sandbox Manager                        ││
│  │  ┌────────────────────────────────────────────────────┐ ││
│  │  │              Sandbox Instance 1                   │ ││
│  │  │  ┌──────────┐ ┌──────────┐ ┌──────────┐          │ ││
│  │  │  │ Tool A   │ │ Tool B   │ │ Tool C   │          │ ││
│  │  │  └──────────┘ └──────────┘ └──────────┘          │ ││
│  │  └────────────────────────────────────────────────────┘ ││
│  │  ┌────────────────────────────────────────────────────┐ ││
│  │  │              Sandbox Instance 2                   │ ││
│  │  │  ┌──────────┐ ┌──────────┐                       │ ││
│  │  │  │ Tool D   │ │ Tool E   │                       │ ││
│  │  │  └──────────┘ └──────────┘                       │ ││
│  │  └────────────────────────────────────────────────────┘ ││
│  └─────────────────────────────────────────────────────────┘│
#### 沙箱隔离机制

OpenClaw的沙箱采用多层隔离策略:

**进程隔离**
工具在独立的子进程或容器中执行,与主进程完全隔离。即使工具崩溃,也不会影响主系统的稳定性。

```python
class ProcessSandbox:
    """进程级沙箱"""
    
    def __init__(self, config: SandboxConfig):
        self.config = config
        self.process_pool = ProcessPoolExecutor(max_workers=config.max_workers)
    
    async def execute(self, tool: Tool, params: dict) -> ToolResult:
        """在隔离进程中执行工具"""
        future = self.process_pool.submit(
            self._run_in_sandbox,
            tool,
            params,
            self.config
        )
        
        try:
            result = await asyncio.wrap_future(future, timeout=self.config.timeout)
            return result
        except TimeoutError:
            self._kill_process(future)
            return ToolResult.error("Tool execution timeout")
        except Exception as e:
            return ToolResult.error(f"Sandbox error: {str(e)}")
    
    def _run_in_sandbox(self, tool: Tool, params: dict, config: SandboxConfig) -> ToolResult:
        """沙箱内部执行"""
        # 设置资源限制
        self._set_limits(config)
        
        # 执行工具
        try:
            return tool.execute(params)
        finally:
            # 清理环境
            self._cleanup()
    
    def _set_limits(self, config: SandboxConfig):
        """设置资源限制"""
        import resource
        
        # 内存限制
        if config.max_memory:
            resource.setrlimit(resource.RLIMIT_AS, (config.max_memory, config.max_memory))
        
        # CPU时间限制
        if config.max_cpu_time:
            resource.setrlimit(resource.RLIMIT_CPU, (config.max_cpu_time, config.max_cpu_time))
        
        # 文件描述符限制
        if config.max_fds:
            resource.setrlimit(resource.RLIMIT_NOFILE, (config.max_fds, config.max_fds))

资源限制

沙箱对工具执行施加严格的资源限制:

sandbox_config:
  max_memory: 512MB        # 最大内存使用
  max_cpu_time: 30s        # 最大CPU时间
  max_wall_time: 60s       # 最大执行时间
  max_fds: 100             # 最大文件描述符
  max_file_size: 10MB      # 最大文件大小
  network_access: false    # 是否允许网络访问
  allowed_paths:           # 允许访问的路径
    - "/tmp/sandbox"
    - "/var/openclaw/data"

安全策略

沙箱执行遵循安全策略配置:

class SecurityPolicy:
    """安全策略"""
    
    def check_permission(self, tool: Tool, action: str, resource: str) -> bool:
        """检查工具执行权限"""
        # 检查工具是否有权限执行此操作
        tool_permissions = tool.permissions
        
        # 检查用户是否有权限使用此工具
        user_permissions = self.context.user_permissions
        
        # 检查资源访问权限
        resource_permissions = self._get_resource_permissions(resource)
        
        return all([
            action in tool_permissions,
            tool.name in user_permissions,
            self._check_resource_access(resource, resource_permissions)
        ])
    
    def apply_policy(self, execution: ToolExecution):
        """应用安全策略到执行过程"""
        # 记录执行审计日志
        self._log_audit(execution)
        
        # 应用内容过滤
        if self.config.content_filter:
            execution.params = self._filter_content(execution.params)
        
        # 应用数据脱敏
        if self.config.data_masking:
            execution.result = self._mask_sensitive_data(execution.result)

### 3.4 结果封装与返回

工具执行完成后,系统会对结果进行封装和格式化,确保返回给LLM的信息清晰、结构化。

#### 结果封装结构

```python
@dataclass
class ToolResult:
    """工具执行结果"""
    success: bool
    data: Any
    error: Optional[str] = None
    metadata: dict = field(default_factory=dict)
    
    @classmethod
    def ok(cls, data: Any, **metadata) -> "ToolResult":
        """创建成功结果"""
        return cls(success=True, data=data, metadata=metadata)
    
    @classmethod
    def error(cls, message: str, **metadata) -> "ToolResult":
        """创建错误结果"""
        return cls(success=False, data=None, error=message, metadata=metadata)
    
    def to_llm_format(self) -> str:
        """转换为LLM可理解的格式"""
        if self.success:
            if isinstance(self.data, str):
                return self.data
            return json.dumps(self.data, ensure_ascii=False, indent=2)
        else:
            return f"Error: {self.error}"
结果处理流程
class ResultProcessor:
    """结果处理器"""
    
    def process(self, result: ToolResult, tool_def: ToolDefinition) -> ProcessedResult:
        """处理工具执行结果"""
        # 1. 格式化结果
        formatted = self._format_result(result)
        
        # 2. 截断过大结果
        truncated = self._truncate_if_needed(formatted)
        
        # 3. 脱敏敏感信息
        masked = self._mask_sensitive(truncated, tool_def)
        
        # 4. 添加元信息
        enriched = self._enrich_metadata(masked, result.metadata)
        
        return ProcessedResult(
            content=enriched,
            is_truncated=truncated != formatted,
            metadata=result.metadata
        )
    
    def _format_result(self, result: ToolResult) -> str:
        """格式化结果"""
        if not result.success:
            return f"Tool execution failed: {result.error}"
        
        if result.data is None:
            return "Tool executed successfully with no output."
        
        if isinstance(result.data, str):
            return result.data
        
        # 格式化结构化数据
        return json.dumps(result.data, ensure_ascii=False, indent=2)
    
    def _truncate_if_needed(self, content: str) -> str:
        """截断过长内容"""
        max_length = self.config.max_result_length  # 默认10000字符
        
        if len(content) <= max_length:
            return content
        
        truncated = content[:max_length]
        return f"{truncated}\n\n... [Result truncated, showing first {max_length} characters]"
    
    def _mask_sensitive(self, content: str, tool_def: ToolDefinition) -> str:
        """脱敏敏感信息"""
        # 根据工具定义的敏感字段配置进行脱敏
        sensitive_fields = tool_def.sensitive_fields
        
        for field in sensitive_fields:
            # 使用正则表达式匹配并替换
            pattern = rf'("{field}"\s*:\s*")[^"]*(")'
            content = re.sub(pattern, r'\1***\2', content)
        
        return content
返回给LLM

处理后的结果以特定格式返回给LLM:

def build_tool_result_message(tool_call_id: str, result: ProcessedResult) -> dict:
    """构建工具结果消息"""
    return {
        "role": "tool",
        "tool_call_id": tool_call_id,
        "content": result.content
    }

完整的工具调用流程:

1. 用户发送消息
   ↓
2. LLM分析并决定调用工具
   ↓
3. 系统解析工具调用请求
   ↓
4. 参数校验和预处理
   ↓
5. 权限检查
   ↓
6. 在沙箱中执行工具
   ↓
7. 结果封装和处理
   ↓
8. 返回结果给LLM
   ↓
9. LLM整合结果生成最终响应

第四章:内置工具详解

OpenClaw提供了丰富的内置工具集,涵盖文件操作、Shell命令、HTTP请求、浏览器自动化等常见场景。这些工具经过精心设计和安全加固,可以直接在生产环境中使用。

4.1 文件操作工具(fs工具集)

文件操作工具集提供了安全受限的文件系统访问能力。

fs_read - 读取文件
{
  "name": "fs_read",
  "description": "读取指定文件的内容。支持文本文件和二进制文件的读取。",
  "parameters": {
    "type": "object",
    "properties": {
      "path": {
        "type": "string",
        "description": "文件路径,必须是允许访问目录内的绝对路径"
      },
      "encoding": {
        "type": "string",
        "default": "utf-8",
        "description": "文件编码,默认UTF-8"
      },
      "limit": {
        "type": "integer",
        "description": "读取行数限制,不指定则读取全部"
      }
    },
    "required": ["path"]
  }
}

使用示例:

# LLM生成调用
{
  "tool": "fs_read",
  "parameters": {
    "path": "/var/openclaw/data/config.yaml",
    "encoding": "utf-8"
  }
}

# 返回结果
{
  "success": true,
  "data": {
    "path": "/var/openclaw/data/config.yaml",
    "content": "server:\n  port: 8080\n  host: localhost\n...",
    "size": 1024,
    "lines": 50
  }
}
fs_write - 写入文件
{
  "name": "fs_write",
  "description": "向指定文件写入内容。如果文件存在则覆盖,不存在则创建。",
  "parameters": {
    "type": "object",
    "properties": {
      "path": {
        "type": "string",
        "description": "文件路径"
      },
      "content": {
        "type": "string",
        "description": "要写入的内容"
      },
      "mode": {
        "type": "string",
        "enum": ["write", "append"],
        "default": "write",
        "description": "写入模式:write覆盖,append追加"
      }
    },
    "required": ["path", "content"]
  }
}
fs_list - 列出目录
{
  "name": "fs_list",
  "description": "列出指定目录下的文件和子目录",
  "parameters": {
    "type": "object",
    "properties": {
      "path": {
        "type": "string",
        "description": "目录路径"
      },
      "pattern": {
        "type": "string",
        "description": "文件名匹配模式,支持通配符"
      },
      "recursive": {
        "type": "boolean",
        "default": false,
        "description": "是否递归列出子目录"
      }
    },
    "required": ["path"]
  }
}
安全限制

文件操作工具实施严格的安全限制:

class FileSystemSecurity:
    """文件系统安全控制"""
    
    def __init__(self, config: FSConfig):
        self.allowed_paths = config.allowed_paths
        self.denied_patterns = config.denied_patterns
        self.max_file_size = config.max_file_size
    
    def validate_path(self, path: str, operation: str) -> ValidationResult:
        """验证路径是否允许访问"""
        abs_path = os.path.abspath(path)
        
        # 检查是否在允许的目录内
        if not any(abs_path.startswith(allowed) for allowed in self.allowed_paths):
            return ValidationResult.error(f"Path '{path}' is outside allowed directories")
        
        # 检查是否匹配拒绝模式
        for pattern in self.denied_patterns:
            if re.match(pattern, abs_path):
                return ValidationResult.error(f"Path '{path}' matches denied pattern")
        
        # 检查文件大小(对于读取操作)
        if operation == "read" and os.path.exists(abs_path):
            size = os.path.getsize(abs_path)
            if size > self.max_file_size:
                return ValidationResult.error(f"File size {size} exceeds maximum {self.max_file_size}")
        
        return ValidationResult.ok()

### 4.2 Shell命令工具

Shell命令工具允许执行系统命令,但受到严格的安全限制。

#### shell_execute - 执行命令

```json
{
  "name": "shell_execute",
  "description": "在沙箱环境中执行Shell命令。仅允许预定义的安全命令。",
  "parameters": {
    "type": "object",
    "properties": {
      "command": {
        "type": "string",
        "description": "要执行的命令"
      },
      "timeout": {
        "type": "integer",
        "default": 30,
        "description": "执行超时时间(秒)"
      }
    },
    "required": ["command"]
  }
}

安全命令白名单

ALLOWED_COMMANDS = {
    # 文本处理
    "grep", "sed", "awk", "cut", "sort", "uniq", "wc",
    # 文件操作
    "ls", "find", "cat", "head", "tail", "stat",
    # 网络诊断
    "ping", "curl", "wget",
    # 系统信息
    "date", "uptime", "df", "du", "free"
}

DENIED_PATTERNS = [
    r"rm\s+-rf",           # 危险删除
    r">\s*/dev/",          # 设备写入
    r"\|\s*sh",            # 管道执行
    r";\s*rm",             # 命令注入
    r"\$\([^)]+\)",        # 命令替换
]

4.3 HTTP请求工具

HTTP请求工具用于与外部API和Web服务交互。

http_request - 发送HTTP请求
{
  "name": "http_request",
  "description": "发送HTTP请求到指定URL。支持GET、POST、PUT、DELETE等方法。",
  "parameters": {
    "type": "object",
    "properties": {
      "url": {
        "type": "string",
        "format": "uri",
        "description": "请求URL"
      },
      "method": {
        "type": "string",
        "enum": ["GET", "POST", "PUT", "DELETE", "PATCH"],
        "default": "GET"
      },
      "headers": {
        "type": "object",
        "description": "请求头"
      },
      "body": {
        "type": "string",
        "description": "请求体"
      },
      "timeout": {
        "type": "integer",
        "default": 30,
        "description": "超时时间(秒)"
      }
    },
    "required": ["url"]
  }
}

安全限制

class HTTPSecurityPolicy:
    """HTTP请求安全策略"""
    
    ALLOWED_DOMAINS = [
        "api.openai.com",
        "api.anthropic.com",
        "*.googleapis.com"
    ]
    
    DENIED_DOMAINS = [
        "localhost",
        "127.0.0.1",
        "10.*", "172.16.*", "192.168.*"
    ]
    
    MAX_REDIRECTS = 5
    MAX_RESPONSE_SIZE = 10 * 1024 * 1024  # 10MB

4.4 Browser浏览器工具

Browser工具提供Web自动化能力,用于网页抓取和交互。

browser_navigate - 导航页面
{
  "name": "browser_navigate",
  "description": "导航到指定URL并加载页面",
  "parameters": {
    "type": "object",
    "properties": {
      "url": {
        "type": "string",
        "format": "uri",
        "description": "目标URL"
      },
      "wait_until": {
        "type": "string",
        "enum": ["load", "domcontentloaded", "networkidle"],
        "default": "load"
      }
    },
    "required": ["url"]
  }
}
browser_screenshot - 截图
{
  "name": "browser_screenshot",
  "description": "截取当前页面的屏幕截图",
  "parameters": {
    "type": "object",
    "properties": {
      "full_page": {
        "type": "boolean",
        "default": false,
        "description": "是否截取完整页面"
      },
      "selector": {
        "type": "string",
        "description": "CSS选择器,截取特定元素"
      }
    }
  }
}

4.5 其他常用工具

数据库工具
{
  "name": "db_query",
  "description": "执行SQL查询",
  "parameters": {
    "type": "object",
    "properties": {
      "query": {
        "type": "string",
        "description": "SQL查询语句"
      },
      "database": {
        "type": "string",
        "description": "数据库名称"
      }
    },
    "required": ["query"]
  }
}
邮件工具
{
  "name": "email_send",
  "description": "发送电子邮件",
  "parameters": {
    "type": "object",
    "properties": {
      "to": {
        "type": "array",
        "items": {"type": "string"},
        "description": "收件人列表"
      },
      "subject": {
        "type": "string",
        "description": "邮件主题"
      },
      "body": {
        "type": "string",
        "description": "邮件内容"
      }
    },
    "required": ["to", "subject", "body"]
  }
}

第五章:自定义工具开发

OpenClaw提供了完善的工具开发框架,开发者可以轻松创建自定义工具来扩展系统能力。

5.1 工具开发流程

开发一个自定义工具需要经过以下步骤:

1. 设计工具接口
   ├── 确定工具功能
   ├── 定义输入参数
   └── 定义输出格式

2. 实现工具类
   ├── 继承BaseTool类
   ├── 实现execute方法
   └── 添加错误处理

3. 编写工具定义
   ├── JSON Schema定义
   ├── 描述和示例
   └── 权限配置

4. 测试与调试
   ├── 单元测试
   ├── 集成测试
   └── 安全测试

5. 注册与发布
   ├── 本地注册
   ├── 配置部署
   └── 共享发布

5.2 TypeScript工具实现

以下是一个完整的自定义工具实现示例:

// tools/WeatherTool.ts
import { BaseTool, ToolResult, ToolDefinition } from '@openclaw/core';
import axios from 'axios';

interface WeatherParams {
  city: string;
  unit?: 'celsius' | 'fahrenheit';
}

interface WeatherData {
  temperature: number;
  condition: string;
  humidity: number;
  wind: string;
}

export class WeatherTool extends BaseTool<WeatherParams, WeatherData> {
  readonly name = 'get_weather';
  readonly description = '查询指定城市的当前天气信息';

  getDefinition(): ToolDefinition {
    return {
      name: this.name,
      description: this.description,
      parameters: {
        type: 'object',
        properties: {
          city: {
            type: 'string',
            description: '城市名称,如"北京"、"上海"'
          },
          unit: {
            type: 'string',
            enum: ['celsius', 'fahrenheit'],
            default: 'celsius',
            description: '温度单位'
          }
        },
        required: ['city']
      }
    };
  }

  async execute(params: WeatherParams): Promise<ToolResult<WeatherData>> {
    try {
      // 调用天气API
      const response = await axios.get(
        `https://api.weather.com/v1/current`,
        {
          params: {
            city: params.city,
            unit: params.unit || 'celsius'
          },
          timeout: 10000
        }
      );

      const data = response.data;

      return ToolResult.ok({
        temperature: data.temp,
        condition: data.condition,
        humidity: data.humidity,
        wind: data.wind
      });
    } catch (error) {
      return ToolResult.error(
        `天气查询失败: ${error.message}`
      );
    }
  }
}

5.3 工具测试与调试

// tests/WeatherTool.test.ts
import { WeatherTool } from '../tools/WeatherTool';

describe('WeatherTool', () => {
  let tool: WeatherTool;

  beforeEach(() => {
    tool = new WeatherTool();
  });

  test('工具定义正确', () => {
    const def = tool.getDefinition();
    expect(def.name).toBe('get_weather');
    expect(def.parameters.required).toContain('city');
  });

  test('成功查询天气', async () => {
    const result = await tool.execute({ city: '北京' });
    expect(result.success).toBe(true);
    expect(result.data).toHaveProperty('temperature');
  });

  test('处理无效城市', async () => {
    const result = await tool.execute({ city: '不存在的城市' });
    expect(result.success).toBe(false);
  });
});

5.4 工具发布与共享

打包配置

{
  "name": "@openclaw/tool-weather",
  "version": "1.0.0",
  "main": "dist/index.js",
  "types": "dist/index.d.ts",
  "peerDependencies": {
    "@openclaw/core": "^2.0.0"
  },
  "keywords": ["openclaw", "tool", "weather"],
  "openclaw": {
    "tools": ["WeatherTool"]
  }
}

发布到npm

npm publish --access public

第六章:Tool Calling最佳实践

基于OpenClaw的实践经验,我们总结了一套Tool Calling的最佳实践指南,帮助开发者设计高质量的AI Agent工具。

6.1 工具命名与描述

命名规范

✅ 好的命名:
- get_weather      (动词_名词)
- send_email       (动词_名词)
- db_query         (领域_动作)

❌ 不好的命名:
- weather          (缺少动作)
- doSomething      (模糊)
- tool1            (无意义)

描述技巧

{
  "description": "查询指定城市的当前天气信息,包括温度、湿度、风速、天气状况等。适用于用户询问天气、出行建议等场景。仅支持中国主要城市。"
}

6.2 参数设计原则

必填vs可选

{
  "properties": {
    "query": {
      "type": "string",
      "description": "搜索关键词(必填)"
    },
    "limit": {
      "type": "integer",
      "default": 10,
      "description": "返回结果数量(可选,默认10)"
    }
  },
  "required": ["query"]
}

枚举优于字符串

{
  "unit": {
    "type": "string",
    "enum": ["celsius", "fahrenheit"],
    "description": "温度单位"
  }
}

6.3 错误处理策略

class RobustTool(BaseTool):
    async def execute(self, params):
        try:
            result = await self._do_work(params)
            return ToolResult.ok(result)
        except ValidationError as e:
            return ToolResult.error(f"参数错误: {e}")
        except TimeoutError:
            return ToolResult.error("操作超时,请稍后重试")
        except PermissionError:
            return ToolResult.error("权限不足")
        except Exception as e:
            self.logger.exception("工具执行异常")
            return ToolResult.error(f"内部错误: {type(e).__name__}")

6.4 性能优化技巧

1. 异步执行

async def execute_parallel(self, calls):
    tasks = [self.execute(call) for call in calls]
    return await asyncio.gather(*tasks)

2. 结果缓存

@cache(ttl=300, key="weather:{city}")
async def get_weather(self, city: str):
    ...

3. 超时控制

result = await asyncio.wait_for(
    self._execute_internal(params),
    timeout=self.config.timeout
)

本章小结

本章深入探讨了OpenClaw的Tool Calling工具调用机制,从基础概念到实现细节,涵盖了完整的技术栈:

核心收获

  1. Tool Calling基础:理解了工具调用的工作原理和发展历程,认识到这是AI Agent从"对话"走向"行动"的关键技术。

  2. 工具定义规范:掌握了JSON Schema工具定义方法,学会了如何设计清晰、规范的工具接口。

  3. 执行流程:了解了从LLM决策到结果返回的完整执行链路,包括参数校验、沙箱隔离、结果封装等关键环节。

  4. 内置工具:熟悉了OpenClaw提供的文件操作、Shell命令、HTTP请求、浏览器自动化等内置工具的使用方法。

  5. 自定义开发:学会了如何开发和发布自定义工具,扩展OpenClaw的能力边界。

  6. 最佳实践:掌握了工具命名、参数设计、错误处理、性能优化的实用技巧。

关键设计理念

  • 安全优先:所有工具调用都在沙箱中执行,确保系统安全
  • 标准化:采用业界主流规范,确保与各类LLM的兼容性
  • 可扩展:灵活的注册和发现机制,支持工具生态的持续扩展

下篇预告

在下一篇博客中,我们将探讨对话管理与上下文工程。这是AI Agent系统的"大脑",负责管理对话状态、维护上下文信息、协调各个模块的协作。

下篇将涵盖以下内容:

  1. 对话状态管理:如何跟踪和维护多轮对话的状态
  2. 上下文压缩:在有限的Token预算内保留关键信息
  3. 记忆系统:短期记忆与长期记忆的设计与实现
  4. 多会话管理:支持用户同时进行多个独立对话
  5. 对话分支与回溯:支持对话历史的管理和恢复

敬请期待!

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐