OpenClaw 技能开发决策报告：脚本内置分析逻辑 vs. 框架原生调用
OpenClaw 技能开发决策报告：脚本内置分析逻辑 vs. 框架原生调用
背景
针对大批量数据处理任务（如医疗条目标准化、2000+ 日志分析），探讨 Skill 文件的最佳构建模式，明确在 Skill 脚本中直接封装 API 调用逻辑的必要性。
核心模式定义
模式 A：脚本内置分析逻辑（Fat Skill）
Skill 的 Python 脚本中直接编写大模型 API 请求代码（httpx/openai库），在脚本内完成数据解析与回写。
角色：Skill 是一个闭环的任务处理器。
[Agent 触发] → [Skill 内部: 读取数据 → 并发调用 LLM → 校验 → 写入 DB] → [返回统计摘要]模式 B：框架原生调用（Lean Skill）
Skill 脚本仅负责原子化动作（读取一行数据、执行一个浏览器点击），分析逻辑交给 OpenClaw 主 Agent 按 ReAct 循环处理。
角色：Skill 是 Agent 的感官辅助。
[Agent 思考] → [调用 Skill 获取数据] → [Agent 分析] → [调用 Skill 写入] → [Agent 思考] → ...维度对比
| 维度 | 模式 A（Fat Skill） | 模式 B（Lean Skill） |
|---|---|---|
| 吞吐量 | 极高，Python 异步并发处理数百条 | 低，受限于 Agent 串行步进 |
| Token 消耗 | 精简，脚本预清洗后仅发送核心字段 | 高，Agent 携带大量上下文 |
| 容错能力 | 强，局部重试 + 独立超时控制 | 弱，单步失败常导致 Task 挂起 |
| 输出一致性 | 高，代码控制格式（JSON/HTML） | 中，受模型指令遵循能力影响 |
| 扩展性 | 极佳，可直接触发 DB 写入 | 一般，需等 Agent 返回再处理 |
| 开发成本 | 较高，需编写完整处理逻辑 | 低，复用框架能力 |
| 调试便利性 | 好，可独立运行脚本测试 | 差，需在 Agent 链路中观察 |
为什么大批量处理需要"逻辑下沉"
1. 绕过 ReAct 决策负担
OpenClaw Agent 每次调用模型前都要进行"自我对话"（Thought）。对于批量任务，这种思考是纯粹的资源浪费。
- 2000 条数据 × 每条 1 次 Thought ≈ 2000 次无意义的决策开销
- 脚本内置方案直接跳过决策，转为批处理
2. 解决超时风险
框架层的 30s 超时是全局性的，无法按任务特点调整。
# 脚本内置：独立超时 + 分段处理asyncdefanalyze_batch(items,timeout_per_item=10):semaphore=asyncio.Semaphore(20)# 并发控制asyncdefprocess_one(item):asyncwithsemaphore:try:returnawaitasyncio.wait_for(call_llm(item),timeout=timeout_per_item)exceptasyncio.TimeoutError:return{"status":"timeout","item_id":item["id"]}returnawaitasyncio.gather(*[process_one(i)foriinitems])3. 数据预清洗
医疗日志中常有大量重复堆栈信息。脚本在 API 请求前通过正则过滤无效信息，可节省 40%+ Token 成本。
importredefprune_log(raw_log:str)->str:"""清洗日志，去除重复堆栈和无效字段"""# 去除重复堆栈帧raw_log=re.sub(r'(at .+\n)\1+',r'\1',raw_log)# 去除时间戳噪音raw_log=re.sub(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+Z\s*','',raw_log)# 截断过长内容returnraw_log[:2000]4. 成本对比估算
以 2000 条医疗日志分析为例：
| 指标 | 模式 A | 模式 B |
|---|---|---|
| 总 API 调用次数 | ~2000（仅分析） | ~6000（Thought + Action + Observation） |
| 平均每条 Token | ~500（清洗后） | ~1500（含上下文） |
| 总 Token 消耗 | ~100 万 | ~900 万 |
| 预计耗时 | 5-10 分钟（并发 20） | 3-5 小时（串行） |
| 失败恢复 | 仅重试失败条目 | 需从断点人工重启 |
推荐架构：混合技能模式
┌─────────────────────────────────────────┐ │ OpenClaw Agent │ │ （负责任务调度、异常上报、结果汇总） │ └──────────────────┬──────────────────────┘ │ 触发 ▼ ┌─────────────────────────────────────────┐ │ Fat Skill（批处理技能） │ │ │ │ 1. 读取数据源（DB / 文件 / API） │ │ 2. 预清洗 & 分组 │ │ 3. asyncio 并发调用 LLM │ │ 4. 结果校验 & 标准化 │ │ 5. 持久化写入 MySQL / ES │ │ 6. 返回统计摘要给 Agent │ └─────────────────────────────────────────┘完整示例
importasyncioimportjsonfromopenaiimportAsyncOpenAI client=AsyncOpenAI()asyncdefbatch_analyze_skill(task_config:dict)->dict:"""批量分析技能入口"""# 1. 读取数据items=awaitload_items(task_config["source"])# 2. 预清洗cleaned=[prune_log(item["content"])foriteminitems]# 3. 并发分析semaphore=asyncio.Semaphore(task_config.get("concurrency",20))results={"success":[],"failed":[]}asyncdefanalyze_one(idx,content):asyncwithsemaphore:try:resp=awaitasyncio.wait_for(client.chat.completions.create(model="gpt-4o-mini",messages=[{"role":"system","content":task_config["prompt"]},{"role":"user","content":content}],response_format={"type":"json_object"}),timeout=15)parsed=json.loads(resp.choices[0].message.content)results["success"].append({"id":items[idx]["id"],"data":parsed})exceptExceptionase:results["failed"].append({"id":items[idx]["id"],"error":str(e)})awaitasyncio.gather(*[analyze_one(i,c)fori,cinenumerate(cleaned)])# 4. 写入数据库awaitsave_results(results["success"])# 5. 返回摘要return{"total":len(items),"success":len(results["success"]),"failed":len(results["failed"]),"failed_ids":[r["id"]forrinresults["failed"]]}场景决策矩阵
| 数据量 | 任务特点 | 推荐方案 |
|---|---|---|
| 1-10 条 | 灵活多变，需 Agent 决策 | 模式 B（Lean Skill） |
| 10-100 条 | 格式半固定，中等效率要求 | 混合：Agent 编排 + 小批量脚本 |
| 100+ 条 | 格式固定，强调效率 | 模式 A（Fat Skill） |
| 医疗/金融级 | 要求数据强一致性回写 | 模式 A（Fat Skill） |
结论
对于搜索失败日志分析及医疗数据标准化项目，采用模式 A（脚本内置分析逻辑）。原因：
- 吞吐量提升 20-50 倍（并发 vs 串行）
- Token 成本降低 80%+（预清洗 + 无 Thought 开销）
- 生产级容错（局部重试、断点续传）
- 输出格式稳定可控（代码约束 > 模型指令遵循）
对于少量、探索性、需要灵活决策的任务，仍应使用模式 B 发挥 Agent 的推理优势。