1. 为什么我最终选了OpenClaw PolarDB这套组合先交代一下背景。我们团队要在企业内部落地一个AI Agent目标非常务实让业务同学用自然语言查数据库、跑统计数据、按时生成报表而不是每次都得提工单等数据组排期。前期我们也试过自己从零搭Agent框架把大模型、工具调用、会话管理、权限体系全部串起来结果发现工作量远超预期。后来转向OpenClaw生态用它的Skills机制加上Flow编排把这件事从三个月规划压缩到了两周出原型并且整体架构仍然保持可控。这篇博文就是一次完整复盘重点讲清楚PolarDB Agent Express Skills是怎么开发的Flow编排又是怎么把零散的技能串成业务闭环的。OpenClaw这个框架核心解决的是Agent如何真正执行任务的问题而不是只停留在对话层面。它提供了技能Skills机制让Agent可以调用外部工具、操作数据库、处理文件同时提供了Flow编排能力让多个技能可以按业务逻辑串起来形成自动化流程。PolarDB作为企业数据基座兼容MySQL和PostgreSQL生态云原生架构下扩容、备份、高可用都是现成的非常适合作为Agent的数据服务层。把两者组合起来我就得到了一个能够听懂需求—查数—分析—产出结果的Agent而不是一个只会聊天的机器人。本文出现的完整配置和代码我都基于实际可跑的方案整理过你可以直接照着落地。1.1 企业AI Agent落地到底难在哪先别急着写代码我先把痛点说透。企业里做AI Agent真正的难点从来不是接一个大模型API而是下面这几件事。第一工具调用链路特别长。一个业务同学问上个月华东区的销售额环比增长了多少Agent要先识别意图然后确定要查哪张表、用什么聚合逻辑再去执行SQL最后把结果转成自然语言。这里面每一步都是工程问题不是单纯靠提示词能解决的。第二权限和数据安全不能糊弄。数据库不是随便让Agent裸连的查询必须受控要防注入、防误删、防越权。我们内部有明文要求Agent产生的数据库操作必须可审计、可回滚。第三编排逻辑要灵活。真实业务不是一问一答而是先查A表再根据结果决定是否查B表最后生成报告这种多步流程。如果每加一个环节都要改代码那这个Agent根本没法维护。我选择OpenClaw正是因为它在第二个和第三个问题上给出了相对成熟的解法。Skills可以理解为一种能力封装单元每个Skill负责一个明确的原子能力比如查询PolarDB、生成图表、发送钉钉消息Flow则负责把这些Skill按业务逻辑连起来分支、循环、超时、重试都有对应的处理手段。1.2 PolarDB当数据基座优势不只是兼容数据底座这块我也犹豫过要不要直接用自建MySQL或者PostgreSQL。后来还是定了PolarDB原因有三。第一PolarDB对MySQL和PostgreSQL的高度兼容意味着团队迁移成本低。我们内部很多历史SQL脚本可以直接复用Agent在生成SQL时模型的训练语料里关于MySQL/PostgreSQL的语法记忆也能直接生效生成准确率会比方言数据库高不少。第二PolarDB的弹性能力对Agent这类读多写少、查询负载突发的场景很友好。Agent一旦被业务同学大量使用查询并发会突然冲高PolarDB的Serverless形态可以自动扩缩容不至于半夜一个报表任务就把数据库打挂。第三企业级能力齐全。跨可用区高可用、备份恢复、SQL审计这些能力都是开箱即用的这在过安全合规评审的时候非常加分。简单说我需要的不是一个能连上的数据库而是一个敢让Agent去操作的企业级数据服务。PolarDB在这个维度上是达标的。1.3 这套组合最终落在哪些业务场景我们实际跑通的场景主要有三类你也可以对照自己的需求看是否匹配。第一类是自然语言查数。业务同学在Agent里输入查一下最近7天订单失败率最高的10个城市Agent通过Skills解析语义、生成SQL、查询PolarDB然后把结果以表格或摘要形式返回。这类场景对Skill的要求是SQL生成可控、查询结果格式化。第二类是周期性报表生成。比如每天早晨9点自动拉取前一天的经营数据计算环比、同比生成一份Markdown格式的日报并推送到群里面。这类场景靠单个Skill不够必须靠Flow编排把查数—计算—生成文档—推送串起来。第三类是辅助分析。比如给出一份订单明细让Agent按多个维度做聚合分析输出数据洞察。这类场景考验的是Skill对数据结构的理解能力和Flow对多分支处理的支持。2. Skills开发给Agent造一双数据库专用手2.1 先理解OpenClaw的Skill机制在OpenClaw生态里Skill是Agent能力的最小执行单元。我习惯用一句话来理解Skill就是一个带说明书的外挂工具。Agent本身不直接执行SQL而是根据用户的请求决定调用哪个Skill然后把参数传给SkillSkill执行完返回结构化结果Agent再把结果组织成回复。一个Skill通常由三个部分组成。第一部分是清单文件manifest描述这个Skill的名字、用途、参数定义。就像外卖平台的菜单告诉Agent这个Skill能做什么、需要什么料。第二部分是执行代码Claw实现具体的逻辑比如连接数据库、执行SQL、处理异常。这部分可以是你熟悉的任意语言我习惯用Python生态成熟、用的人多。第三部分是使用说明instructions告诉大模型在什么情况下调用这个Skill、参数怎么填、需要注意什么。这部分非常关键因为Agent是靠说明来决定何时出手的。我在这个项目里做的PolarDB Agent Express Skills本质上就是一种面向PolarDB数据服务的快速技能包。我把它命名为Express意思是开箱即用、参数精简、适合快速组装不追求覆盖所有数据库操作而是把最高频的查询、元数据查看、慢SQL分析这几个动作打磨好。2.2 动手写一个PolarDB查询Skill先放一个我们实际在用的查询Skill示例完整展示清单文件、代码和说明文档。这是我简化后的版本把内部业务逻辑都去掉了但结构保持原样。首先是manifest文件我这里使用YAML格式name: polar_db_query description: 查询PolarDB数据库支持传入只读SQL语句返回结构化查询结果 version: 1.0.0 language: python parameters: - name: sql type: string required: true description: 只读SQL查询语句必须是SELECT开头 - name: table_hint type: string required: false description: 可选的表名提示用于辅助大模型理解查询意图 - name: limit type: integer required: false description: 返回最大行数默认50最大500 default: 50然后是Python执行代码#!/usr/bin/env python3 import argparse import json import os import re import pymysql ALLOWED_PREFIX (select, show, describe, explain) BLOCK_KEYWORDS (insert, update, delete, drop, alter, truncate, grant, revoke) def parse_args(): parser argparse.ArgumentParser() parser.add_argument(--sql, requiredTrue) parser.add_argument(--table_hint, default) parser.add_argument(--limit, typeint, default50) return parser.parse_args() def validate_sql(sql: str, limit: int): sql_stripped sql.strip().rstrip(;) sql_lower sql_stripped.lower() if not sql_lower.startswith(ALLOWED_PREFIX): raise ValueError(SQL必须以SELECT/SHOW/DESCRIBE/EXPLAIN开头) for keyword in BLOCK_KEYWORDS: if re.search(r\b keyword r\b, sql_lower): raise ValueError(fSQL中包含被禁止的关键字: {keyword}) if limit 500: raise ValueError(limit不能超过500) return sql_stripped def execute_query(sql: str, limit: int): conn pymysql.connect( hostos.getenv(POLARDB_HOST), portint(os.getenv(POLARDB_PORT, 3306)), useros.getenv(POLARDB_USER), passwordos.getenv(POLARDB_PASSWORD), databaseos.getenv(POLARDB_DATABASE), connect_timeout5, read_timeout15, charsetutf8mb4, ) try: with conn.cursor() as cursor: cursor.execute(sql) columns [desc[0] for desc in cursor.description] rows cursor.fetchmany(limit) return {columns: columns, rows: rows, row_count: len(rows)} finally: conn.close() def main(): args parse_args() try: sql validate_sql(args.sql, args.limit) result execute_query(sql, args.limit) print(json.dumps(result, ensure_asciiFalse, defaultstr)) except Exception as e: print(json.dumps({error: str(e)}, ensure_asciiFalse)) raise SystemExit(1) if __name__ __main__: main()最后是Skill说明文档这部分决定了Agent在什么场景下会调用它# polar_db_query 使用说明 ## 功能 执行只读SQL查询返回列名和行数据。适用于查数、取数、数据验证等场景。 ## 何时使用 - 用户提问涉及具体数据、统计指标、订单记录等需要查询数据库时 - 用户请求查一下统计一下看看多少等操作时 ## 参数填写规则 - sql必须是以SELECT/SHOW/DESCRIBE/EXPLAIN开头的只读语句 - 不允许出现INSERT/UPDATE/DELETE/DROP/ALTER等写操作关键字 - 如果用户的问题涉及具体时间范围应尽量在SQL中显式写出时间条件 - 如果表名不在常见表中先调用polar_db_list_tables确认表名 ## 注意事项 1. 所有SQL必须带LIMIT避免全表扫描 2. 涉及金额字段时注意单位避免误解 3. 查询结果只返回原始数据不做计算如需计算请使用数据分析Skill 4. 禁止尝试绕过校验规则执行写操作这三件套放在同一个目录下Skill就定义完成了。我把这个Skill放到OpenClaw的skills目录中并注册到Agent的能力列表里Agent就能在合适的时机调用它。2.3 Skill输入校验和安全边界开发Skills时数据库安全是我最关注的部分。上面代码里的校验逻辑看起来简单但每一行都对应着实际踩过的坑。首先SQL前缀白名单校验。只允许SELECT、SHOW、DESCRIBE、EXPLAIN四种只读操作。之所以用前缀匹配而不是正则全局匹配是因为前缀判断简单可靠不容易被绕过。如果大模型生成了一段以SELECT开头但内部包含写操作的SQL后面的关键字拦截会兜底。这种双重校验我建议所有涉及数据库的Skill都做上。其次敏感关键字拦截。这里用正则表达式检测insert、update、delete、drop等词。需要注意单纯字符串包含判断会在列名或注释中出现误判比如一个列名恰好叫updated_at就会误伤。所以这里用\b词边界匹配同时把updated放在前面拦截掉避免updated_at漏网。我的经验是宁可错杀也不能放过企业对写操作是零容忍的。再次连接信息通过环境变量注入。数据库密码、用户名不出现在代码和Skill配置里而是放在运行环境的环境变量中。这样既能避免密钥泄露也方便在部署时根据环境切换不同的连接配置。如果你是企业场景我强烈建议再把密钥托管到KMS之类的服务里。最后设置连接和读取超时。数据库连接超时设为5秒读取超时设为15秒避免Agent卡在某个慢查询上不释放连接。这个参数看着不起眼但在并发情况下能救命。我们线上就出现过一次慢SQL把连接池打满的事故后来所有数据库Skill都强制加了超时。2.4 Skills本地调试与部署流程再讲讲调试。我建议开发阶段不要直接让Agent去调用Skill而是先用命令行本地跑一遍。以我们的polar_db_query为例。第一步设置环境变量export POLARDB_HOSTpc-xxxxx.polarDB.polardb.aliyuncs.com export POLARDB_PORT3306 export POLARDB_USERagent_reader export POLARDB_PASSWORDyour_password export POLARDB_DATABASEsales_db第二步手动执行测试参数校验和查询逻辑python polar_db_query.py --sql SELECT region, SUM(amount) FROM sales WHERE dt2025-06-01 GROUP BY region LIMIT 10 --limit 10如果这一步返回正常的JSON说明Skill本身没问题。再去OpenClaw里加载测试就能把Agent问题和Skill问题分开排查。我见过很多同事上来就让Agent调Skill出错了根本不知道是Agent选错了参数还是Skill代码写错了白白浪费时间。部署方面OpenClaw支持将多个Skill放入指定目录统一管理。我习惯按能力域建子目录比如database/、analysis/、report/每个目录下放一个独立Skill。这样后续新增Skill不会互相干扰也方便给前端同学单独授权某些能力。3. Flow编排把Skill串成周报自动生成流水线单个Skill解决的是点的问题到了Flow编排这一层解决的是线和面的问题。如果Agent只能调用单个Skill那它充其量是个高级命令行工具。但有了Flow编排它才真正变成一条完整的自动化业务链路。我拿我们最常用的经营数据周报自动生成来说完整梳理一遍Flow是怎么设计的。3.1 编排层到底做了什么Flow编排的核心是把一个复杂的业务目标拆解成多个步骤并定义步骤之间的流转关系。这有点像工厂里的流水线每个工位只干一件专精的事但物料在不同工位之间流转最终组装出成品。OpenClaw的Flow里每个节点可以对应一个Skill调用也可以对应一段内置逻辑比如读取配置条件判断数据转换。节点之间通过字段传递数据前一个节点的输出可以作为后一个节点的输入。这种设计的好处是每个节点都很简单但组合起来可以处理相当复杂的业务。我在实际项目里总结了一条经验Flow节点要设计成单一职责。不要试图让一个节点既查数又算环比又生成图表而是拆成多个节点。这样做的好处是出错时能快速定位到底是哪个环节挂了也方便以后替换某个节点。比如今天用PolarDB明天想换成其他数据库只需要替换查询节点后面的分析节点完全不用动。3.2 周报任务的整体链路设计周报这个场景我们拆成下面这几个节点。节点1参数初始化。读取周报配置包括统计周期本周一到周日、需要统计的业务线、接收人列表。节点2调用polar_db_query查询本周订单总量、销售额、活跃用户数等核心指标。节点3调用polar_db_query查询上周同期的对应数据用于环比计算。节点4调用数据分析Skill计算环比增长率、找出Top5增长品类和Top5下跌品类。节点5调用报表生成Skill把数据和分析结论整理成结构化的Markdown报告。节点6调用消息推送Skill把报告发送到企业微信群或钉钉群。这里有个设计细节值得说明为什么把查询本周数据和查询上周数据拆成两个节点而不是在一个SQL里用子查询搞定因为Agent生成的复杂SQL出错的概率会显著增加一旦某个表名写错或JOIN逻辑不对整个Flow都会失败。而拆成两个简单查询每个查询都容易校验即使其中之一失败也能单独重试不至于把整个流程推翻重来。3.3 关键节点的配置示例在Flow的配置中每个节点的定义大致长这样。我用一个接近实际配置的YAML片段来示意nodes: - id: init type: config config: biz_lines: [零售, 电商, 企业服务] stats_cycle: 本周 next: query_current - id: query_current type: skill skill: polar_db_query input: sql: SELECT biz_line, SUM(order_amount) AS total_amount, COUNT(DISTINCT user_id) AS active_users FROM orders WHERE dt BETWEEN {start_date} AND {end_date} GROUP BY biz_line table_hint: orders, 订单表包含biz_line业务线、order_amount订单金额、user_id用户ID、dt日期字段 limit: 50 next: query_previous - id: query_previous type: skill skill: polar_db_query input: sql: SELECT biz_line, SUM(order_amount) AS total_amount, COUNT(DISTINCT user_id) AS active_users FROM orders WHERE dt BETWEEN {prev_start_date} AND {prev_end_date} GROUP BY biz_line table_hint: orders, 订单表同上 limit: 50 next: calc_analysis - id: calc_analysis type: skill skill: data_analysis input: current_data: ${query_current.output} previous_data: ${query_previous.output} next: gen_report - id: gen_report type: skill skill: report_generator input: analysis_result: ${calc_analysis.output} template: weekly_report_template.md next: push_message - id: push_message type: skill skill: messenger_push input: message: ${gen_report.output} target: wecom_group需要注意里面${query_current.output}这种引用方式它表示把前一个节点的输出作为当前节点的参数传入。这种数据流设计是整个Flow的黏合剂。配置好之后Flow引擎会按照next字段依次执行节点并把数据正确地传给下一个节点。3.4 状态、超时、失败与幂等Flow一旦跑起来就不再是一次对话那么简单它会涉及状态管理和异常处理。这里面我有几条实操经验。第一每个Flow执行最好有独立的执行ID。这个ID贯穿日志、数据库操作、消息推送全链路。出了问题运营同学直接把执行ID发给我们我们就能在日志系统里一键定位到某个具体节点而不是去人肉翻聊天记录。第二为每个节点设置超时时间。PolarDB查询节点我们设15秒超时报表生成节点设30秒消息推送设10秒。超时后Flow引擎要能识别并触发重试或失败分支。我们的策略是数据库查询超时后最多重试2次推送失败则直接走告警分支通知管理员。第三具备幂等性。这个问题容易忽略。如果Flow因为网络抖动被重复执行会不会重复推送消息会不会重复计算数据我们给每个Flow执行加了唯一请求ID推送节点做去重判断同一请求ID在同一时间段内只允许推送一次。这样即使上游重试下游也不会重复打扰用户。第四设计失败补偿分支。比如查询本周数据失败我们不会傻等而是让流程进入异常分支重新尝试连接备用只读节点。如果备用节点也失败就把失败信息发送给运维群并保留中间数据等人工介入处理。4. 常见问题排查与避坑实录这块是我最想写的内容。整个项目做下来踩过的坑比顺利跑通的部分更有价值。我按环境、数据库、Flow执行三个维度整理成速查表。4.1 OpenClaw环境安装与加载问题第一个高频报错是Windows下安装时出现could not safely verify the WSL2 environment。这个报错的意思是OpenClaw在启动时检查WSL2环境发现虚拟机相关的配置不满足要求。原因通常是Windows Subsystem for Linux版本过旧或者电脑没有开启虚拟化功能。排查顺序是先确认Windows版本是否支持WSL2然后执行wsl --update把内核升级到最新最后要确保控制面板里虚拟机平台和适用于Linux的Windows子系统两个功能都勾选启用。我帮同事处理过一台机器一开始怎么装都不行后来发现是公司域策略禁用了虚拟化功能和IT沟通开放权限后马上解决了。如果你确认能跑通用的Linux容器但OpenClaw仍报这个错那大概率是双系统或老版本WSL残留导致的重装WSL内核后重启一次基本就好了。第二个常见问题是Skills目录加了新Skill但Agent始终调用不到。这种问题80%出在清单文件没写好。比如参数定义缺少required字段或者description写得太模糊。我见过一个同事把description写成query the databaseAgent根本不知道这个Skill适合查什么数据自然就不会调用。建议description里明确写出查询订单数据、用户数据、销售数据这类具体信息大模型才会在合适的场景想起它。第三个问题是Skill代码改了但Agent行为没变化。这种情况通常是缓存问题。OpenClaw对Skill清单有缓存机制不是每次请求都重新读取磁盘。解决办法是在管理界面或命令行里手动刷新Skill列表或者直接重启OpenClaw服务。我在开发阶段会先本地命令行测试Skill脚本确认无误后再刷新服务省去反复重启的麻烦。4.2 PolarDB连接与查询常见故障PolarDB连接不上集中在几个原因上。第一是白名单未配置。PolarDB的集群默认只允许白名单内的IP访问。Agent服务部署在K8s集群或函数计算上时出口IP可能不固定需要将服务所在网段加入白名单。如果你发现本地连接正常、线上连接失败基本就是这个问题。第二是SSL/TLS握手失败。PolarDB的默认连接通常需要启用SSL如果代码里没加SSL参数或者证书没配好会在连接阶段报错。我们最后统一在数据库连接串里启用SSL加verify模式并把证书文件放在远程连接密钥目录中让所有Skill共用一个连接配置模块。第三是事务和时区问题。Agent自动生成的SQL如果涉及日期过滤条件务必确认时区设置。我们线上PolarDB默认时区是UTC8但Agent在大模型生成SQL时可能使用了NOW()或CURRENT_TIMESTAMP如果连接会话时区配置不一致就会出现查今天查不到数据这种诡异问题。我的做法是在建立连接时显式执行SET time_zone 08:00避免隐式时区导致的数据偏差。再来谈查询性能。Agent生成的SQL很多时候不会主动带索引条件。比如用户问查一下所有订单Agent可能真的生成SELECT * FROM orders直接把库拖垮。所以Skill层面我做了强制约束不允许没有WHERE条件的查询虽然这会降低灵活性但能保护数据库。我们实际线上还配置了PolarDB的慢SQL分析和限流告警一旦出现执行时间超过2秒的查询会自动通知数据组复核防止Agent生成的逻辑异常SQL产生连锁影响。4.3 Flow执行中的逻辑与异常排查Flow编排最常见的坑是隐形死循环。比如条件分支写得不严谨节点A处理完把结果又传回节点A导致流程反复执行同一个查询。我排查过一例业务同学反馈某次周报跑了两个小时还没出结果看日志发现数据清洗节点被执行了40多次每次都在重复处理同一批数据。原因就是清洗逻辑里有个规则把数据重新标记为未清洗然后又被调度器捞起来处理。第二个坑是并发重复执行。定时任务触发用户手动触发可能同时进行。如果没有做好幂等数据库里会出现两批相同的数据。我们通过唯一请求ID去重表解决每个Flow第一次执行时生成请求ID写入去重表后续相同ID的请求直接忽略。第三个坑是大模型节点输出格式不稳定。Flow里有些节点是大模型直接参与的比如分析数据亮点。大模型输出偶尔会带Markdown符号或换行符导致下游报表生成节点解析失败。我的处理办法是在节点间增加一个格式清洗步骤用简单的Python代码对大模型输出做结构化约束比如强制输出JSON再统一转成标准化数据结构。这一步看似多余实际上让整个Flow的稳定性提高了不少。我整理了一个问题排查速查表你可以直接保存下来问题现象可能原因排查与解决WSL2环境校验失败系统虚拟化未开启/WSL版本过旧开启虚拟机平台功能执行wsl --update更新内核Skill被Agent无视manifest描述不清晰重写description明确适用范围和典型调用场景数据库读取超时SQL缺少谓词/没有LIMIT检查连接配置开启慢SQL审计在Skill层强制加LIMIT定时任务重复执行幂等控制缺失引入唯一请求ID下游按ID去重生成报表格式错乱大模型输出非结构化节点间增加格式清洗强制JSON结构化输出5. 一点个人总结与延伸建议整个项目做下来我最大的感触是OpenClaw PolarDB这套组合确实把企业Agent落地的门槛降低了不少。但不代表它是银弹。Skills和Flow只是给了你一套做好工具化的框架真正决定Agent好用不好用的还是你对业务场景的理解深度和边界控制的严格程度。Skill不要追求大而全要小而精Flow不要追求一次编排解决所有问题要分段迭代安全校验永远是第一优先级永远不要相信大模型的输出。如果你也想在企业里落地类似的Agent我建议从极小的场景开始比如让Agent只做某一张表的TopN查询跑通一个Skill之后再加第二个、第三个最后再考虑Flow编排。千万不要一上来就编排一个十节点的大流程那样出了问题你连定位都困难。再分享一个我现在养成的习惯每个Skill代码里都加上完整注释和README每个Flow节点都保留详细日志。这些东西在开发时觉得费事但上线两个月后你会感谢当时的自己。随着Skills数量增长还建议在OpenClaw里做一次Skill命名规范比如统一用数据域_动作的格式sales_query、sales_report避免agent在选择技能时出现歧义。最后想说AI Agent的能力边界其实是靠人定义的。你给它好用的Skill它就能帮你处理复杂的业务你做好了Flow编排和异常兜底它就能稳定地每天干活。这套OpenClaw生态的实践我们目前还在不断迭代后续如果有新的心得我再回来补充。
1. 为什么我最终选了OpenClaw PolarDB这套组合先交代一下背景。我们团队要在企业内部落地一个AI Agent目标非常务实让业务同学用自然语言查数据库、跑统计数据、按时生成报表而不是每次都得提工单等数据组排期。前期我们也试过自己从零搭Agent框架把大模型、工具调用、会话管理、权限体系全部串起来结果发现工作量远超预期。后来转向OpenClaw生态用它的Skills机制加上Flow编排把这件事从三个月规划压缩到了两周出原型并且整体架构仍然保持可控。这篇博文就是一次完整复盘重点讲清楚PolarDB Agent Express Skills是怎么开发的Flow编排又是怎么把零散的技能串成业务闭环的。OpenClaw这个框架核心解决的是Agent如何真正执行任务的问题而不是只停留在对话层面。它提供了技能Skills机制让Agent可以调用外部工具、操作数据库、处理文件同时提供了Flow编排能力让多个技能可以按业务逻辑串起来形成自动化流程。PolarDB作为企业数据基座兼容MySQL和PostgreSQL生态云原生架构下扩容、备份、高可用都是现成的非常适合作为Agent的数据服务层。把两者组合起来我就得到了一个能够听懂需求—查数—分析—产出结果的Agent而不是一个只会聊天的机器人。本文出现的完整配置和代码我都基于实际可跑的方案整理过你可以直接照着落地。1.1 企业AI Agent落地到底难在哪先别急着写代码我先把痛点说透。企业里做AI Agent真正的难点从来不是接一个大模型API而是下面这几件事。第一工具调用链路特别长。一个业务同学问上个月华东区的销售额环比增长了多少Agent要先识别意图然后确定要查哪张表、用什么聚合逻辑再去执行SQL最后把结果转成自然语言。这里面每一步都是工程问题不是单纯靠提示词能解决的。第二权限和数据安全不能糊弄。数据库不是随便让Agent裸连的查询必须受控要防注入、防误删、防越权。我们内部有明文要求Agent产生的数据库操作必须可审计、可回滚。第三编排逻辑要灵活。真实业务不是一问一答而是先查A表再根据结果决定是否查B表最后生成报告这种多步流程。如果每加一个环节都要改代码那这个Agent根本没法维护。我选择OpenClaw正是因为它在第二个和第三个问题上给出了相对成熟的解法。Skills可以理解为一种能力封装单元每个Skill负责一个明确的原子能力比如查询PolarDB、生成图表、发送钉钉消息Flow则负责把这些Skill按业务逻辑连起来分支、循环、超时、重试都有对应的处理手段。1.2 PolarDB当数据基座优势不只是兼容数据底座这块我也犹豫过要不要直接用自建MySQL或者PostgreSQL。后来还是定了PolarDB原因有三。第一PolarDB对MySQL和PostgreSQL的高度兼容意味着团队迁移成本低。我们内部很多历史SQL脚本可以直接复用Agent在生成SQL时模型的训练语料里关于MySQL/PostgreSQL的语法记忆也能直接生效生成准确率会比方言数据库高不少。第二PolarDB的弹性能力对Agent这类读多写少、查询负载突发的场景很友好。Agent一旦被业务同学大量使用查询并发会突然冲高PolarDB的Serverless形态可以自动扩缩容不至于半夜一个报表任务就把数据库打挂。第三企业级能力齐全。跨可用区高可用、备份恢复、SQL审计这些能力都是开箱即用的这在过安全合规评审的时候非常加分。简单说我需要的不是一个能连上的数据库而是一个敢让Agent去操作的企业级数据服务。PolarDB在这个维度上是达标的。1.3 这套组合最终落在哪些业务场景我们实际跑通的场景主要有三类你也可以对照自己的需求看是否匹配。第一类是自然语言查数。业务同学在Agent里输入查一下最近7天订单失败率最高的10个城市Agent通过Skills解析语义、生成SQL、查询PolarDB然后把结果以表格或摘要形式返回。这类场景对Skill的要求是SQL生成可控、查询结果格式化。第二类是周期性报表生成。比如每天早晨9点自动拉取前一天的经营数据计算环比、同比生成一份Markdown格式的日报并推送到群里面。这类场景靠单个Skill不够必须靠Flow编排把查数—计算—生成文档—推送串起来。第三类是辅助分析。比如给出一份订单明细让Agent按多个维度做聚合分析输出数据洞察。这类场景考验的是Skill对数据结构的理解能力和Flow对多分支处理的支持。2. Skills开发给Agent造一双数据库专用手2.1 先理解OpenClaw的Skill机制在OpenClaw生态里Skill是Agent能力的最小执行单元。我习惯用一句话来理解Skill就是一个带说明书的外挂工具。Agent本身不直接执行SQL而是根据用户的请求决定调用哪个Skill然后把参数传给SkillSkill执行完返回结构化结果Agent再把结果组织成回复。一个Skill通常由三个部分组成。第一部分是清单文件manifest描述这个Skill的名字、用途、参数定义。就像外卖平台的菜单告诉Agent这个Skill能做什么、需要什么料。第二部分是执行代码Claw实现具体的逻辑比如连接数据库、执行SQL、处理异常。这部分可以是你熟悉的任意语言我习惯用Python生态成熟、用的人多。第三部分是使用说明instructions告诉大模型在什么情况下调用这个Skill、参数怎么填、需要注意什么。这部分非常关键因为Agent是靠说明来决定何时出手的。我在这个项目里做的PolarDB Agent Express Skills本质上就是一种面向PolarDB数据服务的快速技能包。我把它命名为Express意思是开箱即用、参数精简、适合快速组装不追求覆盖所有数据库操作而是把最高频的查询、元数据查看、慢SQL分析这几个动作打磨好。2.2 动手写一个PolarDB查询Skill先放一个我们实际在用的查询Skill示例完整展示清单文件、代码和说明文档。这是我简化后的版本把内部业务逻辑都去掉了但结构保持原样。首先是manifest文件我这里使用YAML格式name: polar_db_query description: 查询PolarDB数据库支持传入只读SQL语句返回结构化查询结果 version: 1.0.0 language: python parameters: - name: sql type: string required: true description: 只读SQL查询语句必须是SELECT开头 - name: table_hint type: string required: false description: 可选的表名提示用于辅助大模型理解查询意图 - name: limit type: integer required: false description: 返回最大行数默认50最大500 default: 50然后是Python执行代码#!/usr/bin/env python3 import argparse import json import os import re import pymysql ALLOWED_PREFIX (select, show, describe, explain) BLOCK_KEYWORDS (insert, update, delete, drop, alter, truncate, grant, revoke) def parse_args(): parser argparse.ArgumentParser() parser.add_argument(--sql, requiredTrue) parser.add_argument(--table_hint, default) parser.add_argument(--limit, typeint, default50) return parser.parse_args() def validate_sql(sql: str, limit: int): sql_stripped sql.strip().rstrip(;) sql_lower sql_stripped.lower() if not sql_lower.startswith(ALLOWED_PREFIX): raise ValueError(SQL必须以SELECT/SHOW/DESCRIBE/EXPLAIN开头) for keyword in BLOCK_KEYWORDS: if re.search(r\b keyword r\b, sql_lower): raise ValueError(fSQL中包含被禁止的关键字: {keyword}) if limit 500: raise ValueError(limit不能超过500) return sql_stripped def execute_query(sql: str, limit: int): conn pymysql.connect( hostos.getenv(POLARDB_HOST), portint(os.getenv(POLARDB_PORT, 3306)), useros.getenv(POLARDB_USER), passwordos.getenv(POLARDB_PASSWORD), databaseos.getenv(POLARDB_DATABASE), connect_timeout5, read_timeout15, charsetutf8mb4, ) try: with conn.cursor() as cursor: cursor.execute(sql) columns [desc[0] for desc in cursor.description] rows cursor.fetchmany(limit) return {columns: columns, rows: rows, row_count: len(rows)} finally: conn.close() def main(): args parse_args() try: sql validate_sql(args.sql, args.limit) result execute_query(sql, args.limit) print(json.dumps(result, ensure_asciiFalse, defaultstr)) except Exception as e: print(json.dumps({error: str(e)}, ensure_asciiFalse)) raise SystemExit(1) if __name__ __main__: main()最后是Skill说明文档这部分决定了Agent在什么场景下会调用它# polar_db_query 使用说明 ## 功能 执行只读SQL查询返回列名和行数据。适用于查数、取数、数据验证等场景。 ## 何时使用 - 用户提问涉及具体数据、统计指标、订单记录等需要查询数据库时 - 用户请求查一下统计一下看看多少等操作时 ## 参数填写规则 - sql必须是以SELECT/SHOW/DESCRIBE/EXPLAIN开头的只读语句 - 不允许出现INSERT/UPDATE/DELETE/DROP/ALTER等写操作关键字 - 如果用户的问题涉及具体时间范围应尽量在SQL中显式写出时间条件 - 如果表名不在常见表中先调用polar_db_list_tables确认表名 ## 注意事项 1. 所有SQL必须带LIMIT避免全表扫描 2. 涉及金额字段时注意单位避免误解 3. 查询结果只返回原始数据不做计算如需计算请使用数据分析Skill 4. 禁止尝试绕过校验规则执行写操作这三件套放在同一个目录下Skill就定义完成了。我把这个Skill放到OpenClaw的skills目录中并注册到Agent的能力列表里Agent就能在合适的时机调用它。2.3 Skill输入校验和安全边界开发Skills时数据库安全是我最关注的部分。上面代码里的校验逻辑看起来简单但每一行都对应着实际踩过的坑。首先SQL前缀白名单校验。只允许SELECT、SHOW、DESCRIBE、EXPLAIN四种只读操作。之所以用前缀匹配而不是正则全局匹配是因为前缀判断简单可靠不容易被绕过。如果大模型生成了一段以SELECT开头但内部包含写操作的SQL后面的关键字拦截会兜底。这种双重校验我建议所有涉及数据库的Skill都做上。其次敏感关键字拦截。这里用正则表达式检测insert、update、delete、drop等词。需要注意单纯字符串包含判断会在列名或注释中出现误判比如一个列名恰好叫updated_at就会误伤。所以这里用\b词边界匹配同时把updated放在前面拦截掉避免updated_at漏网。我的经验是宁可错杀也不能放过企业对写操作是零容忍的。再次连接信息通过环境变量注入。数据库密码、用户名不出现在代码和Skill配置里而是放在运行环境的环境变量中。这样既能避免密钥泄露也方便在部署时根据环境切换不同的连接配置。如果你是企业场景我强烈建议再把密钥托管到KMS之类的服务里。最后设置连接和读取超时。数据库连接超时设为5秒读取超时设为15秒避免Agent卡在某个慢查询上不释放连接。这个参数看着不起眼但在并发情况下能救命。我们线上就出现过一次慢SQL把连接池打满的事故后来所有数据库Skill都强制加了超时。2.4 Skills本地调试与部署流程再讲讲调试。我建议开发阶段不要直接让Agent去调用Skill而是先用命令行本地跑一遍。以我们的polar_db_query为例。第一步设置环境变量export POLARDB_HOSTpc-xxxxx.polarDB.polardb.aliyuncs.com export POLARDB_PORT3306 export POLARDB_USERagent_reader export POLARDB_PASSWORDyour_password export POLARDB_DATABASEsales_db第二步手动执行测试参数校验和查询逻辑python polar_db_query.py --sql SELECT region, SUM(amount) FROM sales WHERE dt2025-06-01 GROUP BY region LIMIT 10 --limit 10如果这一步返回正常的JSON说明Skill本身没问题。再去OpenClaw里加载测试就能把Agent问题和Skill问题分开排查。我见过很多同事上来就让Agent调Skill出错了根本不知道是Agent选错了参数还是Skill代码写错了白白浪费时间。部署方面OpenClaw支持将多个Skill放入指定目录统一管理。我习惯按能力域建子目录比如database/、analysis/、report/每个目录下放一个独立Skill。这样后续新增Skill不会互相干扰也方便给前端同学单独授权某些能力。3. Flow编排把Skill串成周报自动生成流水线单个Skill解决的是点的问题到了Flow编排这一层解决的是线和面的问题。如果Agent只能调用单个Skill那它充其量是个高级命令行工具。但有了Flow编排它才真正变成一条完整的自动化业务链路。我拿我们最常用的经营数据周报自动生成来说完整梳理一遍Flow是怎么设计的。3.1 编排层到底做了什么Flow编排的核心是把一个复杂的业务目标拆解成多个步骤并定义步骤之间的流转关系。这有点像工厂里的流水线每个工位只干一件专精的事但物料在不同工位之间流转最终组装出成品。OpenClaw的Flow里每个节点可以对应一个Skill调用也可以对应一段内置逻辑比如读取配置条件判断数据转换。节点之间通过字段传递数据前一个节点的输出可以作为后一个节点的输入。这种设计的好处是每个节点都很简单但组合起来可以处理相当复杂的业务。我在实际项目里总结了一条经验Flow节点要设计成单一职责。不要试图让一个节点既查数又算环比又生成图表而是拆成多个节点。这样做的好处是出错时能快速定位到底是哪个环节挂了也方便以后替换某个节点。比如今天用PolarDB明天想换成其他数据库只需要替换查询节点后面的分析节点完全不用动。3.2 周报任务的整体链路设计周报这个场景我们拆成下面这几个节点。节点1参数初始化。读取周报配置包括统计周期本周一到周日、需要统计的业务线、接收人列表。节点2调用polar_db_query查询本周订单总量、销售额、活跃用户数等核心指标。节点3调用polar_db_query查询上周同期的对应数据用于环比计算。节点4调用数据分析Skill计算环比增长率、找出Top5增长品类和Top5下跌品类。节点5调用报表生成Skill把数据和分析结论整理成结构化的Markdown报告。节点6调用消息推送Skill把报告发送到企业微信群或钉钉群。这里有个设计细节值得说明为什么把查询本周数据和查询上周数据拆成两个节点而不是在一个SQL里用子查询搞定因为Agent生成的复杂SQL出错的概率会显著增加一旦某个表名写错或JOIN逻辑不对整个Flow都会失败。而拆成两个简单查询每个查询都容易校验即使其中之一失败也能单独重试不至于把整个流程推翻重来。3.3 关键节点的配置示例在Flow的配置中每个节点的定义大致长这样。我用一个接近实际配置的YAML片段来示意nodes: - id: init type: config config: biz_lines: [零售, 电商, 企业服务] stats_cycle: 本周 next: query_current - id: query_current type: skill skill: polar_db_query input: sql: SELECT biz_line, SUM(order_amount) AS total_amount, COUNT(DISTINCT user_id) AS active_users FROM orders WHERE dt BETWEEN {start_date} AND {end_date} GROUP BY biz_line table_hint: orders, 订单表包含biz_line业务线、order_amount订单金额、user_id用户ID、dt日期字段 limit: 50 next: query_previous - id: query_previous type: skill skill: polar_db_query input: sql: SELECT biz_line, SUM(order_amount) AS total_amount, COUNT(DISTINCT user_id) AS active_users FROM orders WHERE dt BETWEEN {prev_start_date} AND {prev_end_date} GROUP BY biz_line table_hint: orders, 订单表同上 limit: 50 next: calc_analysis - id: calc_analysis type: skill skill: data_analysis input: current_data: ${query_current.output} previous_data: ${query_previous.output} next: gen_report - id: gen_report type: skill skill: report_generator input: analysis_result: ${calc_analysis.output} template: weekly_report_template.md next: push_message - id: push_message type: skill skill: messenger_push input: message: ${gen_report.output} target: wecom_group需要注意里面${query_current.output}这种引用方式它表示把前一个节点的输出作为当前节点的参数传入。这种数据流设计是整个Flow的黏合剂。配置好之后Flow引擎会按照next字段依次执行节点并把数据正确地传给下一个节点。3.4 状态、超时、失败与幂等Flow一旦跑起来就不再是一次对话那么简单它会涉及状态管理和异常处理。这里面我有几条实操经验。第一每个Flow执行最好有独立的执行ID。这个ID贯穿日志、数据库操作、消息推送全链路。出了问题运营同学直接把执行ID发给我们我们就能在日志系统里一键定位到某个具体节点而不是去人肉翻聊天记录。第二为每个节点设置超时时间。PolarDB查询节点我们设15秒超时报表生成节点设30秒消息推送设10秒。超时后Flow引擎要能识别并触发重试或失败分支。我们的策略是数据库查询超时后最多重试2次推送失败则直接走告警分支通知管理员。第三具备幂等性。这个问题容易忽略。如果Flow因为网络抖动被重复执行会不会重复推送消息会不会重复计算数据我们给每个Flow执行加了唯一请求ID推送节点做去重判断同一请求ID在同一时间段内只允许推送一次。这样即使上游重试下游也不会重复打扰用户。第四设计失败补偿分支。比如查询本周数据失败我们不会傻等而是让流程进入异常分支重新尝试连接备用只读节点。如果备用节点也失败就把失败信息发送给运维群并保留中间数据等人工介入处理。4. 常见问题排查与避坑实录这块是我最想写的内容。整个项目做下来踩过的坑比顺利跑通的部分更有价值。我按环境、数据库、Flow执行三个维度整理成速查表。4.1 OpenClaw环境安装与加载问题第一个高频报错是Windows下安装时出现could not safely verify the WSL2 environment。这个报错的意思是OpenClaw在启动时检查WSL2环境发现虚拟机相关的配置不满足要求。原因通常是Windows Subsystem for Linux版本过旧或者电脑没有开启虚拟化功能。排查顺序是先确认Windows版本是否支持WSL2然后执行wsl --update把内核升级到最新最后要确保控制面板里虚拟机平台和适用于Linux的Windows子系统两个功能都勾选启用。我帮同事处理过一台机器一开始怎么装都不行后来发现是公司域策略禁用了虚拟化功能和IT沟通开放权限后马上解决了。如果你确认能跑通用的Linux容器但OpenClaw仍报这个错那大概率是双系统或老版本WSL残留导致的重装WSL内核后重启一次基本就好了。第二个常见问题是Skills目录加了新Skill但Agent始终调用不到。这种问题80%出在清单文件没写好。比如参数定义缺少required字段或者description写得太模糊。我见过一个同事把description写成query the databaseAgent根本不知道这个Skill适合查什么数据自然就不会调用。建议description里明确写出查询订单数据、用户数据、销售数据这类具体信息大模型才会在合适的场景想起它。第三个问题是Skill代码改了但Agent行为没变化。这种情况通常是缓存问题。OpenClaw对Skill清单有缓存机制不是每次请求都重新读取磁盘。解决办法是在管理界面或命令行里手动刷新Skill列表或者直接重启OpenClaw服务。我在开发阶段会先本地命令行测试Skill脚本确认无误后再刷新服务省去反复重启的麻烦。4.2 PolarDB连接与查询常见故障PolarDB连接不上集中在几个原因上。第一是白名单未配置。PolarDB的集群默认只允许白名单内的IP访问。Agent服务部署在K8s集群或函数计算上时出口IP可能不固定需要将服务所在网段加入白名单。如果你发现本地连接正常、线上连接失败基本就是这个问题。第二是SSL/TLS握手失败。PolarDB的默认连接通常需要启用SSL如果代码里没加SSL参数或者证书没配好会在连接阶段报错。我们最后统一在数据库连接串里启用SSL加verify模式并把证书文件放在远程连接密钥目录中让所有Skill共用一个连接配置模块。第三是事务和时区问题。Agent自动生成的SQL如果涉及日期过滤条件务必确认时区设置。我们线上PolarDB默认时区是UTC8但Agent在大模型生成SQL时可能使用了NOW()或CURRENT_TIMESTAMP如果连接会话时区配置不一致就会出现查今天查不到数据这种诡异问题。我的做法是在建立连接时显式执行SET time_zone 08:00避免隐式时区导致的数据偏差。再来谈查询性能。Agent生成的SQL很多时候不会主动带索引条件。比如用户问查一下所有订单Agent可能真的生成SELECT * FROM orders直接把库拖垮。所以Skill层面我做了强制约束不允许没有WHERE条件的查询虽然这会降低灵活性但能保护数据库。我们实际线上还配置了PolarDB的慢SQL分析和限流告警一旦出现执行时间超过2秒的查询会自动通知数据组复核防止Agent生成的逻辑异常SQL产生连锁影响。4.3 Flow执行中的逻辑与异常排查Flow编排最常见的坑是隐形死循环。比如条件分支写得不严谨节点A处理完把结果又传回节点A导致流程反复执行同一个查询。我排查过一例业务同学反馈某次周报跑了两个小时还没出结果看日志发现数据清洗节点被执行了40多次每次都在重复处理同一批数据。原因就是清洗逻辑里有个规则把数据重新标记为未清洗然后又被调度器捞起来处理。第二个坑是并发重复执行。定时任务触发用户手动触发可能同时进行。如果没有做好幂等数据库里会出现两批相同的数据。我们通过唯一请求ID去重表解决每个Flow第一次执行时生成请求ID写入去重表后续相同ID的请求直接忽略。第三个坑是大模型节点输出格式不稳定。Flow里有些节点是大模型直接参与的比如分析数据亮点。大模型输出偶尔会带Markdown符号或换行符导致下游报表生成节点解析失败。我的处理办法是在节点间增加一个格式清洗步骤用简单的Python代码对大模型输出做结构化约束比如强制输出JSON再统一转成标准化数据结构。这一步看似多余实际上让整个Flow的稳定性提高了不少。我整理了一个问题排查速查表你可以直接保存下来问题现象可能原因排查与解决WSL2环境校验失败系统虚拟化未开启/WSL版本过旧开启虚拟机平台功能执行wsl --update更新内核Skill被Agent无视manifest描述不清晰重写description明确适用范围和典型调用场景数据库读取超时SQL缺少谓词/没有LIMIT检查连接配置开启慢SQL审计在Skill层强制加LIMIT定时任务重复执行幂等控制缺失引入唯一请求ID下游按ID去重生成报表格式错乱大模型输出非结构化节点间增加格式清洗强制JSON结构化输出5. 一点个人总结与延伸建议整个项目做下来我最大的感触是OpenClaw PolarDB这套组合确实把企业Agent落地的门槛降低了不少。但不代表它是银弹。Skills和Flow只是给了你一套做好工具化的框架真正决定Agent好用不好用的还是你对业务场景的理解深度和边界控制的严格程度。Skill不要追求大而全要小而精Flow不要追求一次编排解决所有问题要分段迭代安全校验永远是第一优先级永远不要相信大模型的输出。如果你也想在企业里落地类似的Agent我建议从极小的场景开始比如让Agent只做某一张表的TopN查询跑通一个Skill之后再加第二个、第三个最后再考虑Flow编排。千万不要一上来就编排一个十节点的大流程那样出了问题你连定位都困难。再分享一个我现在养成的习惯每个Skill代码里都加上完整注释和README每个Flow节点都保留详细日志。这些东西在开发时觉得费事但上线两个月后你会感谢当时的自己。随着Skills数量增长还建议在OpenClaw里做一次Skill命名规范比如统一用数据域_动作的格式sales_query、sales_report避免agent在选择技能时出现歧义。最后想说AI Agent的能力边界其实是靠人定义的。你给它好用的Skill它就能帮你处理复杂的业务你做好了Flow编排和异常兜底它就能稳定地每天干活。这套OpenClaw生态的实践我们目前还在不断迭代后续如果有新的心得我再回来补充。
