企业官网建设流程全解析

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 思考] → ...

维度对比

为什么大批量处理需要"逻辑下沉"

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 条医疗日志分析为例：

推荐架构：混合技能模式

┌─────────────────────────────────────────┐ │ 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"]]}

场景决策矩阵

结论

对于搜索失败日志分析及医疗数据标准化项目，采用模式 A（脚本内置分析逻辑）。原因：

1. 吞吐量提升 20-50 倍（并发 vs 串行）
2. Token 成本降低 80%+（预清洗 + 无 Thought 开销）
3. 生产级容错（局部重试、断点续传）
4. 输出格式稳定可控（代码约束 > 模型指令遵循）

对于少量、探索性、需要灵活决策的任务，仍应使用模式 B 发挥 Agent 的推理优势。