OpenClaw技能编写实战：从零构建高效自动化工作流

2次阅读

共计 2660 个字符，预计需要花费 7 分钟才能阅读完成。

OpenClaw 是一个自动化工作流平台，通过编写技能（Skill）实现特定任务的自动化处理。比如自动抓取网页数据、定时触发邮件发送、对接第三方 API 等场景。每个技能本质上是一段可复用的代码逻辑，包含输入处理、核心操作和输出返回三部分。

API 调用混乱 ：直接裸写 requests 调用，缺乏统一管理
容错能力差 ：网络波动或服务异常时直接崩溃
性能问题 ：单线程阻塞式调用，耗时过长
可维护性低 ：日志缺失，出问题难以排查

class BaseSkill:
    def __init__(self):
        self.logger = self._init_logger()

    def _init_logger(self):
        logger = logging.getLogger(self.__class__.__name__)
        logger.setLevel(logging.INFO)
        return logger

    def execute(self, input_data):
        """技能主入口"""
        try:
            processed = self._pre_process(input_data)
            result = self._core_operation(processed)
            return self._post_process(result)
        except Exception as e:
            self.logger.error(f"执行失败: {str(e)}", exc_info=True)
            raise

from tenacity import retry, stop_after_attempt, wait_exponential

class APIClient:
    @retry(stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        reraise=True
    )
    def safe_request(self, method, url, **kwargs):
        try:
            resp = requests.request(method, url, **kwargs)
            resp.raise_for_status()
            return resp.json()
        except requests.exceptions.RequestException as e:
            self.logger.warning(f"API 请求失败: {e}")
            raise

def execute_with_fallback(self, input_data):
    try:
        return self.execute(input_data)
    except TemporaryError as e:  # 可恢复错误
        self.logger.warning(f"临时错误，触发降级: {e}")
        return self._fallback_operation(input_data)
    except CriticalError as e:   # 不可恢复错误
        self.logger.critical(f"关键错误: {e}")
        raise
    finally:
        self._cleanup_resources()  # 资源清理

请求批处理 ：将多个独立请求合并为批量接口调用

def batch_process(items, batch_size=50):
    for i in range(0, len(items), batch_size):
        batch = items[i:i+batch_size]
        yield self.api_client.batch_request(batch)

缓存策略 ：对幂等操作使用 Redis 缓存

@cachetools.cached(cache=RedisCache(), key=lambda x: x['id'])
def get_user_profile(user_id):
    return db.query_user(user_id)

并发控制 ：使用线程池处理 IO 密集型任务

from concurrent.futures import ThreadPoolExecutor

def concurrent_run(tasks):
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(process_task, t) for t in tasks]
        return [f.result() for f in futures]

超时陷阱 ：所有外部调用必须设置超时

# 错误示范
requests.get(url)

# 正确做法
requests.get(url, timeout=(3, 10))  # 连接 3 秒，读取 10 秒

内存泄漏 ：及时释放大对象

# 危险代码
results = []
for item in huge_dataset:
    results.append(process(item))

# 优化方案
for item in huge_dataset:
    yield process(item)  # 使用生成器

日志风暴 ：避免高频打印非关键日志

# 反模式
logger.debug(f"Processing {item_id}")  # 每秒数千次

# 正确方式
if random.random() < 0.01:  # 1% 采样率
    logger.debug("采样日志: %s", item_id)

基于示例框架扩展实现：
1. 调用气象 API 获取指定城市天气
2. 添加失败自动重试
3. 实现缓存（相同城市 10 分钟内不重复查询）
4. 支持批量查询多个城市

class WeatherSkill(BaseSkill):
    def __init__(self):
        super().__init__()
        self.api = WeatherAPI()
        self.cache = cachetools.TTLCache(maxsize=100, ttl=600)

    @cachetools.cached(cachetools.TTLCache(maxsize=100, ttl=600))
    def _core_operation(self, city):
        return self.api.get_weather(city)

通过这个完整案例，你可以掌握 OpenClaw 技能开发的核心模式。记住好的技能应该像乐高积木——单一职责、接口明确、易于组装。在实际项目中，先从简单功能开始，逐步叠加异常处理、性能优化等能力，最终形成稳定可靠的生产级技能。

正文完