Agent 工具调用最佳实践：从 OpenClaw 技能系统说起

写在前面：2026 年，AI Agent 工具调用成为企业 AI 落地的核心技术。这篇文章基于 OpenClaw 技能系统的实战经验，详解工具调用的完整设计思路和最佳实践。

---

一、背景：为什么需要工具调用？

1.1 LLM 的局限性

LLM 能做的：

• ✅ 文本生成
• ✅ 知识问答
• ✅ 代码编写
• ✅ 逻辑推理

LLM 不能做的：

• ❌ 访问实时数据（天气、股票）
• ❌ 执行系统命令（部署、重启）
• ❌ 调用外部 API（数据库、消息队列）
• ❌ 操作用户数据（文件、配置）

1.2 工具调用的价值

用户：部署 OpenClaw 到 K8s

❌ 纯 LLM：
"部署 OpenClaw 到 K8s 需要以下步骤：
1. 创建 PVC
2. 创建 Secret
..."
（只能提供建议，无法执行）

✅ Agent + 工具：
"正在部署 OpenClaw 到 K8s...
✓ 创建 PVC: openclaw-data-pvc
✓ 创建 Secret: openclaw-secrets
✓ 部署 Pod: openclaw-gateway
部署成功！Pod Running (1/1)"
（实际执行并返回结果）

1.3 OpenClaw 技能系统

技能数量：15+
调用次数：日均 1000+
成功率：98%+

核心技能：

• k8s-deploy - K8s 部署
• feishu-doc - 飞书文档操作
• web-search - 网络搜索
• minio-manager - MinIO 文件管理
• tts - 语音合成

---

二、工具定义规范

2.1 工具描述文件（SKILL.md）

示例：skills/k8s-deploy/SKILL.md

# k8s-deploy 技能

**用途**: 部署应用到 Kubernetes 集群

**触发条件**:
- 用户提到"部署"、"kubectl"、"K8s"
- 包含应用名称和部署目标

**参数**:
- `app_name`: 应用名称（必填）
- `namespace`: 命名空间（选填，默认 default）
- `replicas`: 副本数（选填，默认 1）
- `image`: 容器镜像（选填）

**执行步骤**:
1. 验证参数
2. 生成 YAML 配置
3. 执行 kubectl apply
4. 验证部署状态
5. 返回结果

**错误处理**:
- 参数缺失 → 提示用户补充
- 部署失败 → 返回错误信息
- 超时 → 重试 3 次后报错

**示例**:
用户："部署 myapp 到 production"
→ app_name=myapp, namespace=production

2.2 工具函数定义

Python 示例：

from typing import Optional, Dict, Any
from pydantic import BaseModel, Field

class K8sDeployInput(BaseModel):
    """K8s 部署输入参数"""
    app_name: str = Field(..., description="应用名称")
    namespace: str = Field(default="default", description="命名空间")
    replicas: int = Field(default=1, description="副本数")
    image: Optional[str] = Field(None, description="容器镜像")

def k8s_deploy(params: K8sDeployInput) -> Dict[str, Any]:
    """部署应用到 K8s"""
    
    # 1. 验证参数
    if not params.app_name:
        return {"success": False, "error": "应用名称不能为空"}
    
    # 2. 生成 YAML
    yaml_content = generate_deployment_yaml(params)
    
    # 3. 执行部署
    try:
        result = exec_kubectl_apply(yaml_content, params.namespace)
    except Exception as e:
        return {"success": False, "error": str(e)}
    
    # 4. 验证状态
    status = wait_for_deployment(params.app_name, params.namespace)
    
    # 5. 返回结果
    return {
        "success": status == "Running",
        "message": f"部署成功：{params.app_name}",
        "status": status,
        "details": {
            "namespace": params.namespace,
            "replicas": params.replicas
        }
    }

2.3 工具注册机制

OpenClaw 技能注册：

# skills/registry.py

class SkillRegistry:
    def __init__(self):
        self.skills = {}
    
    def register(self, name: str, func: callable, schema: dict):
        """注册技能"""
        self.skills[name] = {
            "func": func,
            "schema": schema,
            "enabled": True
        }
    
    def get(self, name: str) -> dict:
        """获取技能"""
        return self.skills.get(name)
    
    def list(self) -> list:
        """列出所有技能"""
        return list(self.skills.keys())

# 注册技能
registry = SkillRegistry()

registry.register(
    name="k8s-deploy",
    func=k8s_deploy,
    schema={
        "type": "object",
        "properties": {
            "app_name": {"type": "string"},
            "namespace": {"type": "string"},
            "replicas": {"type": "integer"},
            "image": {"type": "string"}
        },
        "required": ["app_name"]
    }
)

---

三、工具调用流程

3.1 完整流程

用户请求
  ↓
意图识别（LLM）
  ↓
选择工具（Skill Router）
  ↓
提取参数（Parameter Extractor）
  ↓
参数验证（Validator）
  ↓
执行工具（Executor）
  ↓
结果处理（Result Handler）
  ↓
返回用户

3.2 意图识别

LLM Prompt：

你是一个 AI Agent 助手。请分析用户意图，选择合适的工具。

可用工具：
1. k8s-deploy - 部署应用到 K8s
2. feishu-doc - 操作飞书文档
3. web-search - 网络搜索
4. minio-manager - MinIO 文件管理

用户请求：部署 OpenClaw 到 K8s

请回答：
1. 是否需要调用工具？（是/否）
2. 选择哪个工具？
3. 提取参数（JSON 格式）

示例回答：
{
  "need_tool": true,
  "tool_name": "k8s-deploy",
  "parameters": {
    "app_name": "OpenClaw",
    "namespace": "default"
  }
}

3.3 参数提取

Few-shot 示例：

用户：部署 myapp 到 production，3 个副本
→ {"app_name": "myapp", "namespace": "production", "replicas": 3}

用户：查下北京天气
→ {"city": "北京", "tool": "get_weather"}

用户：把这篇文章保存到飞书
→ {"doc_title": "文章标题", "content": "文章内容", "tool": "feishu-doc"}

3.4 参数验证

Pydantic 验证：

from pydantic import ValidationError

def validate_params(tool_name: str, params: dict) -> tuple:
    """验证参数"""
    
    schema = get_tool_schema(tool_name)
    
    try:
        # 根据工具选择验证模型
        if tool_name == "k8s-deploy":
            validated = K8sDeployInput(**params)
        elif tool_name == "feishu-doc":
            validated = FeishuDocInput(**params)
        else:
            validated = params
        
        return True, validated, None
    
    except ValidationError as e:
        return False, None, str(e)

# 使用示例
success, validated, error = validate_params("k8s-deploy", params)

if not success:
    return f"参数错误：{error}"

---

四、错误处理机制

4.1 错误分类

错误类型	说明	处理策略
参数错误	参数缺失/类型错误	提示用户补充
工具执行失败	API 调用失败	重试 3 次
超时	执行时间过长	异步执行 + 轮询
权限不足	缺少访问权限	提示用户授权
资源不足	系统资源限制	降级或排队

4.2 重试机制

import time
from functools import wraps

def retry(max_attempts=3, delay=1, backoff=2):
    """重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            current_delay = delay
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    
                    log(f"第{attempt+1}次失败：{e}")
                    time.sleep(current_delay)
                    current_delay *= backoff
            
            return None
        return wrapper
    return decorator

# 使用示例
@retry(max_attempts=3, delay=2, backoff=2)
def k8s_deploy(params):
    # 部署逻辑
    pass

4.3 降级策略

def execute_with_fallback(tool_name: str, params: dict):
    """带降级的工具执行"""
    
    # 主方案
    try:
        return execute_primary(tool_name, params)
    except PrimaryError as e:
        log(f"主方案失败：{e}")
    
    # 备用方案 1
    try:
        return execute_fallback_1(tool_name, params)
    except Exception as e:
        log(f"备用方案 1 失败：{e}")
    
    # 备用方案 2：返回友好提示
    return {
        "success": False,
        "message": "暂时无法完成该操作，请稍后重试",
        "suggestion": "您可以尝试手动执行：kubectl apply -f ..."
    }

4.4 错误信息友好化

def format_error_message(error: Exception, tool_name: str) -> str:
    """格式化错误信息"""
    
    error_messages = {
        "ConnectionError": "网络连接失败，请检查网络",
        "TimeoutError": "操作超时，请稍后重试",
        "PermissionError": "权限不足，请联系管理员授权",
        "ResourceExhausted": "资源不足，请释放部分资源",
    }
    
    error_type = type(error).__name__
    
    if error_type in error_messages:
        return error_messages[error_type]
    
    # 默认：返回技术细节（给开发者）
    return f"{tool_name} 执行失败：{str(error)}"

---

五、OpenClaw 实战案例

5.1 案例 1：K8s 部署

用户请求：部署 OpenClaw 到 K8s

执行过程：

步骤 1：意图识别
- 检测到关键词："部署"、"K8s"
- 选择工具：k8s-deploy

步骤 2：参数提取
- app_name: "OpenClaw"
- namespace: "default"（默认值）
- replicas: 1（默认值）

步骤 3：参数验证
- ✓ app_name 存在
- ✓ namespace 有效
- ✓ replicas 在合理范围

步骤 4：执行部署
✓ kubectl apply -f pvc.yaml
✓ kubectl create secret generic openclaw-secrets
✓ kubectl apply -f deployment.yaml
✓ kubectl apply -f service.yaml

步骤 5：验证状态
✓ kubectl get pods → Running (1/1)
✓ kubectl get svc → ClusterIP 分配
✓ curl health check → 200 OK

步骤 6：返回结果
{
  "success": true,
  "message": "OpenClaw 部署成功",
  "details": {
    "pod": "openclaw-gateway-56b6855c5-qjnmd",
    "status": "Running",
    "service": "192.169.218.43:80"
  }
}

5.2 案例 2：飞书文档操作

用户请求：把这篇文章保存到飞书

执行过程：

步骤 1：意图识别
- 检测到关键词："保存"、"飞书"
- 选择工具：feishu-doc

步骤 2：参数提取
- doc_title: "AI Agent 工具调用最佳实践"
- content: "文章内容..."
- folder_token: （从上下文获取）

步骤 3：执行保存
✓ feishu_doc(action="create", title=doc_title, content=content)
→ doc_token: "dscnABC123..."

步骤 4：返回结果
{
  "success": true,
  "message": "文档已保存到飞书",
  "doc_url": "https://feishu.cn/docx/dscnABC123..."
}

5.3 案例 3：多工具协作

用户请求：搜索 RAG 技术文章，保存到飞书，然后生成语音

执行过程：

步骤 1：分解任务
- 任务 1：搜索 RAG 技术文章 → web-search
- 任务 2：保存到飞书 → feishu-doc
- 任务 3：生成语音 → tts

步骤 2：顺序执行
✓ web-search(query="RAG 技术")
  → 搜索结果：10 篇文章

✓ feishu-doc(action="create", content=搜索结果)
  → doc_token: "dscnABC123..."

✓ tts(text=文章摘要)
  → audio_url: "https://.../audio.mp3"

步骤 3：返回结果
{
  "success": true,
  "message": "已完成所有操作",
  "results": {
    "search_count": 10,
    "doc_url": "https://feishu.cn/docx/dscnABC123...",
    "audio_url": "https://.../audio.mp3"
  }
}

---

六、性能优化技巧

6.1 工具缓存

from functools import lru_cache
import hashlib

def cache_tool_result(func):
    """工具结果缓存装饰器"""
    
    cache = {}
    
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 生成缓存 key
        key_data = f"{func.__name__}:{args}:{kwargs}"
        key = hashlib.md5(key_data.encode()).hexdigest()
        
        # 检查缓存
        if key in cache:
            log(f"使用缓存：{key}")
            return cache[key]
        
        # 执行并缓存
        result = func(*args, **kwargs)
        cache[key] = result
        
        return result
    
    return wrapper

# 使用示例
@cache_tool_result
def web_search(query: str) -> list:
    # 搜索逻辑
    pass

6.2 并发执行

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def execute_parallel(tools: list) -> list:
    """并发执行多个工具"""
    
    loop = asyncio.get_event_loop()
    executor = ThreadPoolExecutor(max_workers=10)
    
    # 创建任务
    tasks = [
        loop.run_in_executor(executor, tool_func, params)
        for tool_func, params in tools
    ]
    
    # 并发执行
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    return results

# 使用示例
tools = [
    (web_search, {"query": "RAG 技术"}),
    (web_search, {"query": "AI Agent"}),
    (web_search, {"query": "Function Calling"}),
]

results = await execute_parallel(tools)

6.3 超时控制

import signal
from contextlib import contextmanager

class TimeoutError(Exception):
    pass

@contextmanager
def timeout(seconds: int):
    """超时控制"""
    
    def handler(signum, frame):
        raise TimeoutError(f"执行超时（{seconds}秒）")
    
    # 设置信号处理器
    signal.signal(signal.SIGALRM, handler)
    signal.alarm(seconds)
    
    try:
        yield
    finally:
        signal.alarm(0)

# 使用示例
with timeout(30):
    result = k8s_deploy(params)

---

七、最佳实践清单

7.1 工具设计

• 单一职责 - 每个工具只做一件事
• 参数明确 - 使用 Pydantic 等验证库
• 错误友好 - 返回用户能理解的错误信息
• 日志完整 - 记录执行过程便于调试
• 支持重试 - 网络请求等易失败操作

7.2 安全控制

• 权限验证 - 检查用户是否有权执行
• 参数过滤 - 防止注入攻击
• 资源限制 - 设置 CPU/内存/时间限制
• 审计日志 - 记录所有工具调用

7.3 性能优化

• 结果缓存 - 避免重复执行
• 并发执行 - 独立任务并行处理
• 超时控制 - 防止长时间阻塞
• 降级策略 - 主方案失败有备用方案

7.4 监控告警

• 调用次数 - 监控每个工具的调用频率
• 成功率 - 低于阈值时告警
• 响应时间 - P99 延迟监控
• 错误分布 - 分析常见错误类型

---

八、常见问题

8.1 工具选择不准确

问题：LLM 选择了错误的工具

解决：

1. 优化工具描述（更清晰）
2. 提供更多 Few-shot 示例
3. 添加工具选择置信度阈值

8.2 参数提取失败

问题：LLM 提取的参数不完整

解决：

1. 使用结构化输出（JSON Schema）
2. 添加参数验证和提示
3. 支持多轮对话补充参数

8.3 工具执行慢

问题：某些工具执行时间长

解决：

1. 异步执行 + 轮询结果
2. 添加进度反馈
3. 优化底层实现

---

九、总结

9.1 核心要点

1. 工具定义 - 清晰的 SKILL.md 描述
2. 参数验证 - 使用 Pydantic 等验证库
3. 错误处理 - 重试 + 降级 + 友好提示
4. 性能优化 - 缓存 + 并发 + 超时控制

9.2 行动建议

开发者：

• 从简单工具开始（1-2 个）
• 完善错误处理
• 添加监控和日志

企业：

• 建立工具开发规范
• 统一注册和管理
• 定期审查和优化

---

十、相关链接

• OpenClaw 技能系统：~/workspace/skills/
• Function Calling：https://platform.openai.com/docs/guides/function-calling
• Pydantic：https://docs.pydantic.dev
• 本文代码：~/workspace/obsidian-sync/projects/P3_OpenClaw_Extension/skills/

---

作者：John
创建时间：2026-04-02
文档版本：v1.0