构建一个能真正在企业里跑起来的AI Agent关键并不是把大模型接到数据库这么简单。我最近基于OpenClaw生态做了一组面向PolarDB的Express Skills再用Flow编排串成了完整的企业级Agent流程从慢SQL诊断、索引推荐到故障现场的自动巡检都能让业务同学直接通过自然语言拿结果。这篇文章把整个项目的拆解思路、Skill开发过程、Flow编排细节和踩坑记录都写出来希望给正在做AI Agent落地的团队一些可复用的参考。这套内容适合三类人一类是已经在用LLM API但只停留在问答阶段的开发者一类是负责数据库运维或研发效能想把日常经验整理成自动化能力的工程师还有一类是刚接触Agent架构想知道Skill、Flow这些概念到底该怎么落地的同学。我会尽量少讲空泛概念多给能直接抄走的代码和配置。1. 项目背景与整体设计思路1.1 企业AI Agent为什么需要Skill和Flow过去很多团队接入大模型后做得最多的事情就是在对话窗口里“问一句答一句”。比如业务同学问“最近订单表的慢查询怎么变多了”大模型会给出一些通用建议但下一次遇到类似问题所有工作还得重来一遍。这种模式本质上还是把大模型当搜索引擎用并没有形成企业内部的资产沉淀。OpenClaw这类Agent框架带来的是另一种思路把高频、确定性的能力沉淀成Skill。Skill就是一组可复用的“技能包”里面包含触发条件、输入输出定义、执行脚本、Prompt模板和结果解析逻辑。Flow编排则是把这些Skill串起来的“流程线”决定Agent先做什么、后做什么、什么情况下走分支、什么情况下需要人工介入。我选择OpenClaw作为基座除了它原生支持Skill和Flow之外还有一个很实际的原因它对底层模型是弱绑定的。也就是说团队内部可以根据不同场景切换DeepSeek、通义千问或其他模型而不用重写业务逻辑。对于企业来说这解决了“被单一模型绑架”的顾虑。1.2 PolarDB Express Skills的定位不是大而全而是高频可复用刚开始做PolarDB相关Agent时我犯过一个典型错误想做一个能回答所有数据库问题的巨型Agent。结果Prompt越来越长模型表现越来越不稳定调试成本非常高。后来我调整了策略只做高价值、高频出现的场景并且把它们封装成独立的Express Skills。“Express”这个词在这里的定位是轻量和快速每个Skill只解决一个问题输入输出清晰可以在几分钟内被集成到Flow里。目前我们沉淀的核心Skill包括慢SQL诊断根据慢日志数据输出问题原因、趋势分析和优化建议。索引推荐分析SQL执行计划给出候选索引及预期收益。空间与容量分析查看表空间增长趋势预测扩容时间点。参数配置体检对比当前参数和最佳实践标记风险项。故障现场快照发生连接数突增或锁等待时自动收集关键信息。每个Skill都是独立的可以在不同Flow中复用。比如“慢SQL诊断”既可以单独给DBA用也可以嵌入到每周巡检的自动化Flow里。这种设计让Skill的建设成本被摊薄了越往后复用价值越高。1.3 整体架构控制层、能力层、模型层三层分离整个Agent架构可以拆成三层来看。最上层是控制层负责Flow编排、状态管理、会话记忆和人工审批。中间是能力层也就是那组Express Skills它们负责执行具体的数据库操作或生成分析内容。最底层是模型层和基础设施模型层提供自然语言理解和生成能力基础设施包含PolarDB实例、监控数据源、告警系统等。这种分层最大的好处是“各司其职”。控制层不需要知道每个Skill内部怎么实现只按照Flow定义的路由调用即可能力层不关心用户怎么说只接收结构化的参数并返回结构化的结果模型层则只负责它最擅长的语言理解和生成推理。这样一来任何一层发生变化都不会牵动整体重构。为什么没有把所有逻辑都放在一个Agent Prompt里因为企业级场景对稳定性和可观测性的要求远高于个人玩具项目。Flow编排让每一步执行都有迹可循Skill让每次数据库操作都经过严格校验这套体系才能真正扛住生产环境的压力。2. PolarDB Agent Express Skills 开发实战2.1 Skill的核心组成与编写规范在OpenClaw里一个标准Skill并不是简单的“一段Prompt”而是一个包含元信息、逻辑和资源的完整包。我通常会用下面的目录结构来组织polar_slow_query_skill/ ├── skill.yaml ├── run.py ├── template.py └── requirements.txtskill.yaml是Skill的“身份证”OpenClaw会读取这个文件来决定什么情况下加载和调用这个Skill。下面是我在项目里使用的真实示例name: polar_slow_query_diagnosis description: 分析PolarDB的慢SQL日志定位高耗时SQL并输出优化建议。 trigger: keywords: - 慢查询 - 慢SQL - 响应慢 semantic_match: true input_schema: db_instance: type: string description: PolarDB实例ID required: true time_range: type: string description: 分析时间范围例如 1h、24h、7d default: 1h limit: type: integer description: 返回TOP N条慢SQL default: 10 output_schema: slow_queries: type: array items: sql_id: string sql_text: string execution_count: number avg_latency_ms: number suggestion: string risk_level: low|medium|high这里有几个容易被忽略的细节。semantic_match: true表示除了关键词匹配还会通过模型判断用户语义这样即使用户没说“慢SQL”而是说“最近接口好慢”只要模型判定与慢查询相关Skill也能被触发。input_schema里定义了参数类型和默认值OpenClaw会基于这个Schema自动向用户追问缺失的参数不需要我们自己在代码里做复杂的解析。run.py是Skill的执行逻辑。这个文件必须有明确的主函数和标准的输入输出OpenClaw会以子进程方式调用它。这里我举一个最简版本的实现import json import sys def load_input(): return json.loads(sys.argv[1]) def query_slow_log(instance_id, time_range, limit): # 这里通过PolarDB的OpenAPI或直连只读实例获取慢日志数据 # 返回结构化的慢查询列表 return [] def generate_suggestion(sql_item): # 根据SQL特征和统计信息生成优化建议 return 建议检查该SQL是否缺少合适索引并确认查询条件是否导致全表扫描。 def main(): params load_input() rows query_slow_log(params[db_instance], params[time_range], params[limit]) result [] for item in rows: result.append({ sql_id: item[sql_id], sql_text: item[sql_text], execution_count: item[count], avg_latency_ms: item[avg_latency], suggestion: generate_suggestion(item), risk_level: item[risk_level] }) print(json.dumps(result, ensure_asciiFalse)) if __name__ __main__: main()实际生产环境里我不会在Python脚本中直连数据库写复杂查询而是优先通过PolarDB的OpenAPI取得数据这样可以避免给Agent一个高权限的数据库账号。但无论用什么方式获取数据输出结构一定要和output_schema严格一致。模型看到的是这个结构化结果而不是一堆日志文本输出越规范最终回答的质量就越稳定。2.2 从0开发一个慢SQL诊断Skill以慢SQL诊断这个例子完整跑通一个Skill的开发流程。第一步是明确输入。对于PolarDB来说慢SQL的原始来源是慢日志和性能监控。我们需要拿到实例ID、时间范围、返回条数这三个参数。第二步是获取数据。我习惯写一个独立的query_slow_log函数内部封装对PolarDB OpenAPI的调用如果OpenAPI拉取失败再降级到直连只读实例查询performance_schema中的相关表。第三步是让LLM生成建议。这一步并不是把SQL文本丢给模型就算完而是要把上下文先格式化好。实际开发时我会在Python脚本里先组装一份“分析材料”。例如对一条慢SQL整理出执行次数、平均延迟、最大延迟、扫描行数、返回行数、当前表索引信息然后再调用大模型生成分析建议。这样做比直接让模型“看SQL给建议”准确得多因为模型充分了解了执行环境。下面这个是Prompt模板的核心片段你是一名资深的PolarDB数据库性能优化专家。 请根据以下慢SQL信息输出诊断结论和优化建议。 SQL文本{sql_text} 执行次数{execution_count} 平均耗时{avg_latency_ms}ms 最大耗时{max_latency_ms}ms 扫描行数{rows_examined} 返回行数{rows_sent} 当前索引{index_list} 要求 1. 说明该SQL性能问题的可能原因。 2. 给出具体的优化建议包括索引变更、SQL改写或参数调整。 3. 评估风险等级low/medium/high。这里有一个非常关键的实践在模板里只放与问题相关的结构化数据不要把整个数据库的表结构和所有历史日志都塞进去。Token长度受限是一方面更重要的是上下文越长模型输出的可靠性越差。让Skill先完成数据筛选模型只做精炼分析和决策这才是正确的分工。开发过程中我还会为Skill写一个template.py把最终要展示给用户的Markdown报告模板独立出来。这样后续调整报告样式不需要改动核心的run.py逻辑。模板比如这样### 慢SQL诊断结果{db_instance} | SQL ID | 执行次数 | 平均耗时(ms) | 风险等级 | 建议 | |--------|----------|--------------|----------|------| | {sql_id} | {count} | {avg_latency} | {risk} | {suggestion} |2.3 Skill的测试与本地联调Skill写完以后第一件事不是接到Agent里而是直接用命令行模拟OpenClaw的调用方式进行单测。因为OpenClaw本质上是以进程方式调用Skill所以我们可以用模拟参数直接跑脚本python run.py {db_instance: pc-xxxx, time_range: 24h, limit: 5}运行以后检查输出是否符合预期的JSON结构。我见过很多开发者在Skill还没测试通过的情况下就直接接入Flow结果问题排查起来非常痛苦因为你不确定是Flow配置问题、模型识别问题还是Skill内部逻辑问题。所以我建议把“本地直接执行”作为Skill开发的基本习惯。如果返回结果不符合预期有一个很常见的坑是系统输出的JSON里包含了多余的提示信息。比如有些开发者在print(json.dumps(result))之前又加了一句print(查询成功)。这在命令行里看起来没什么但OpenClaw会解析整个stdout多余的输出会导致JSON解析失败。解决办法是确保run.py的stdout只输出一次最终的JSON对象所有日志都写到stderr。3. Flow编排把独立Skill串成企业级Agent3.1 Flow编排的核心概念与执行模型Skill解决的是“单点能力”Flow解决的才是“复杂任务”。进入Flow编排之前需要理解几个核心概念节点Node、过渡Transition、状态State和人工审批Human-in-the-loop。节点Flow中的每个执行步骤可以是调用Skill、调用LLM、执行代码、等待人工输入。过渡节点之间的连接关系包括条件分支、无条件跳转、超时跳转。状态Flow执行过程中的上下文数据所有节点共享这部分数据。人工审批在某些高风险的数据库操作前暂停执行等待人工确认。在OpenClaw中Flow通常用YAML或JSON文件定义。相比传统代码流程声明式的Flow编排更容易被业务人员理解也方便版本管理。我个人比较喜欢把Flow文件提交到Git仓库里每次修改都走Code Review因为Agent的执行流程直接影响生产数据库必须有严格的变更管控。以下是一个简单的Flow定义示例考虑一个“慢SQL自动诊断”的场景flow: id: polar_slow_query_flow start: parse_request nodes: - id: parse_request type: llm action: name: extract_params prompt: | 从用户请求中提取以下参数 - 数据库实例ID - 时间范围 - 返回条数 返回JSON格式。 next: check_permission - id: check_permission type: code action: name: permission_check run: permission_check.py next: - condition: has_permission to: run_skill - condition: otherwise to: reject - id: run_skill type: skill skill: polar_slow_query_diagnosis params: db_instance: ${parse_request.db_instance} time_range: ${parse_request.time_range} limit: ${parse_request.limit} next: generate_report - id: generate_report type: llm action: name: format_report prompt: | 请根据慢SQL诊断结果生成一份面向业务研发的优化报告。 注意不要使用过多数据库专业术语。 end: true - id: reject type: code action: name: notify_reject run: notify_reject.py end: true在这个Flow里每个节点只做一件明确的事情数据通过${node_name.field_name}从上游节点获取。这样好处非常明显如果某一步出错了我们可以只看对应节点和上下文不需要把整个链路翻一遍。3.2 实战构建“PolarDB健康巡检与优化建议”Agent我实际落地的一个Flow是企业PolarDB实例的周巡检。过去DBA每周要手动执行十几条SQL再花半天时间整理Word报告。现在通过Agent Flow自动化完成并且支持人工抽查。整个Flow执行过程如下定时触发每周一早上九点通过调度平台调用Flow入口。实例清单获取从配置中心读取需要巡检的PolarDB实例列表。批量执行Skill对每个实例并行执行“慢SQL诊断”“空间分析”“参数体检”三个Skill。风险汇总调用LLM把所有实例的结果汇总成一张风险矩阵。生成报告调用报告模板Skill输出Markdown或PDF文件。发送通知将报告推送到企业微信或钉钉群高风险项单独负责人。这个Flow里最需要注意的环节是“批量执行”和“风险汇总”的衔接。如果对每个实例串行执行几十个实例就要跑几个小时并行执行又可能会有PolarDB OpenAPI的并发限制。所以我们在Flow里加了并发度控制通过配置max_concurrency限制同时执行的实例数量。在实际配置里我还会为每个Skill加超时时间。数据库分析类的Skill通常需要几秒到几分钟不等如果超过5分钟还没返回Flow就把这个实例标记为“超时”继续处理下一个最后在报告中单独列出超时项。不要让一个慢节点阻塞整个巡检流程。3.3 条件分支与人工审批的设计Flow中的人工审批节点是企业级Agent非常关键的环节。就拿“索引推荐”这个Skill来说它分析完SQL后可能会建议“在user_id上创建索引”。对于一般表来说这是个低风险操作但如果这个表有数亿行数据直接执行DDL可能会带来锁表和主从延迟风险应该自动进入人工审批。我在Flow里这样设计的Skill返回结果中除了建议还有一个risk_level字段。如果risk_level low直接执行下一步。如果risk_level high进入审批节点发送审批请求给DBA群。DBA审批通过后Flow继续执行变更审批拒绝则记录原因并结束。人工审批的节点在OpenClaw里可以通过回调方式实现。当Flow遇到审批节点时会先输出一个等待状态并把审批URL推到IM工具。DBA点击审批通过后Flow通过Webhook接收结果并继续执行。这里建议所有高风险操作都设置审批超时时间比如24小时未审批则自动挂起并发送提醒防止Flow一直占着资源。4. 常见问题与排查技巧4.1 WSL2环境校验失败的处理开发阶段很多同事遇到“OpenClaw could not safely verify the WSL2 environment”的报错尤其是在Windows开发机上跑本地调试环境。这个报错的原因通常是WSL2内核版本过旧或者OpenClaw无法读取WSL2的配置文件。排查时可以先执行wsl --status确认WSL2已设置为默认版本。然后在PowerShell里用管理员权限运行wsl --update把内核组件更新到最新版本。如果更新后依然报错可以考虑把项目目录从/mnt/c/移动到WSL2的原生文件系统比如~/workspace目录下再重新初始化。这里有一个小技巧OpenClaw需要验证WSL2环境本质上是为了确保容器或沙箱能安全运行。如果你用的是Windows开发机但实际项目部署在Linux服务器上本地开发环境不必强求解决这个报错可以直接在Linux服务器上操作或者在Windows下使用Docker Desktop的WSL2集成模式避免两边折腾。4.2 Skill不生效或LLM不调用指定SkillAgent已经接上了Skill但用户怎么说都不触发这个问题我遇到过很多次。原因通常集中在三个方面第一是Skill注册后没有重新加载。OpenClaw在运行时会缓存Skill列表改完skill.yaml后需要重启Agent进程或执行热加载命令。第二是trigger词典里的关键词和实际用户表达差太远。比如用户说“数据库最近有点卡”你的关键词只有“慢SQL”那自然匹配不到。解决办法是扩充关键词列表同时开启semantic_match。第三是Skill名称和描述过于笼统。如果描述写“数据库分析”LLM拿不准什么时候该用这个Skill。更好的写法是“当用户询问PolarDB慢查询、执行计划、索引优化问题时使用此Skill可基于慢日志输出诊断报告”。描述越具体模型越容易做出正确的路由决策。4.3 Flow执行超时和死循环Flow编排里最常见的故障就是死循环。比如你定义了一个循环节点让Agent在没有拿到足够参数时不断向用户追问但用户没有及时回复Flow就卡在那里了。设计Flow时要给每个节点设置超时时间和最大迭代次数。如果某个节点需要等待用户输入超时后应该走“默认值”分支而不是一直等下去。另外循环节点里建议增加一个“出口条件”比如最多追问三次超过三次就使用默认参数继续执行避免流程卡死。还有一个容易踩的坑是并行节点的资源泄漏。如果Flow里开了多个并行分支某个分支抛异常后其他分支可能仍在运行。最好在Flow引擎层配置统一的全局超时确保整个Flow有明确的生命周期。4.4 数据库操作权限与安全问题在PolarDB场景里Agent一定会碰数据权限控制绝不能省。我的原则是所有Agent使用的数据库账号都是只读账号只授予查询权限不授予写权限。凡是涉及变更的操作建索引、改参数、杀会话都必须走人工审批节点由具备权限的DBA执行。日志审计要记录Agent每一次执行的信息包括用户请求、Skill名称、执行参数、LLM输出方便事后回溯。对PolarDB的OpenAPI调用进行IP白名单限制不要放通所有来源。在这个基础上还可以做一个“危险操作前置识别”的Skill在真正调用数据库管理接口之前先让模型判断当前操作是否可能对生产环境造成影响。如果判断结果为高风险就直接拦截并提示人工介入。多一层保护生产环境的安全系数会高很多。5. 落地过程中的一些体会项目做到这里我个人最大的收获是企业级AI Agent的建设不是“做一个机器人”而是“搭一套有纪律的能力编排系统”。Skill把经验固化Flow把流程固化而OpenClaw这样的框架提供了让这两者高效协作的底座。最初的开发速度可能会慢一些但每沉淀一个Skill后续的Agent能力就会肉眼可见地增强。还有一点经验想分享给刚开始做Agent开发的团队不要把希望完全寄托在模型智商上要把更多精力花在输入输出的结构化和异常处理上。一个边界清晰的Skill配合一个考虑周全的Flow远比一个聪明的Prompt更可靠。数据库类Agent尤其如此因为它面对的是真实业务和真实数据一次失误的代价可能是很大的。如果你也在做类似的事情建议从最痛的一两个场景切入先用Flow把一个简单的Skill串起来跑通再慢慢扩展技能包。技术本身不复杂复杂的是对业务场景的理解和对边界的敬畏。
