confluent-skill-creator

Confluent 技能创建器

专用技能创建器，用于构建针对真实 Confluent 环境（本地、Platform、Cloud 或 WarpStream）进行测试的 Confluent 产品技能。

此技能在基础技能创建器工作流之上，加入了 Confluent 专属要求：

• 灵活的范围：技能可以围绕用例构建（例如，“从 PostgreSQL 到 Iceberg 的 CDC”），也可以针对特定组件（例如，“Flink SQL 技能”、“Schema Registry 技能”）
• 平台定位：技能要么针对特定平台（在技能中注明），要么通过特定平台的参考文件支持所有平台
• 环境测试：针对本地 Confluent、Confluent Platform、Confluent Cloud 或 WarpStream 运行真实的端到端测试
• 安全的 CRUD 操作：在创建或修改 Confluent 资源前，展示完整计划并征求确认
• 凭据管理：为所有必需的凭据正确设置 .env 文件
• 规范合规性：根据 Agent Skills 规范验证创建的技能
• 智能默认值：默认使用 JSON_SR 的 Schema Registry（在标头中包含架构 GUID）

前置条件

• Python 3.9+
• confluent-kafka[avro,json,protobuf]、requests、jsonschema、python-dotenv Python 包（参见 scripts/requirements.txt）
• 可访问 Confluent 环境（Cloud、Platform、本地 Docker 或 WarpStream）以进行端到端测试

何时使用此技能

当用户想要为外部用户创建涉及以下内容的技能时，请使用此技能：

• Kafka 主题、生产者、消费者或流处理应用程序
• Flink SQL 或流处理
• 连接器（Debezium CDC、接收器连接器等）
• Schema Registry（Avro、JSON Schema、Protobuf）
• 用于数据湖集成的 Tableflow
• 从数据库到数据湖的 CDC 管道
• 实时数据丰富或转换

核心工作流

1. 了解范围和平台定位（Confluent 专属）
2. 收集需求并进行访谈
3. 编写技能初稿
4. 验证规范合规性（Confluent 专属）
5. 设置测试环境和凭据（Confluent 专属）
6. 创建测试用例并运行端到端测试（Confluent 专属）
7. 使用查看器评估
8. 迭代直至满意
9. 优化描述
10. 打包并交付

---

技能结构与渐进式披露

将这些结构原则应用于你创建的每项技能。它们扩展了基础的 技能创建器指南；第 4 步中的规范合规性检查会强制执行具体要求，而本节则提供思维模型。

技能结构

skill-name/
├── SKILL.md (required)
│   ├── YAML frontmatter (name, description required)
│   └── Markdown instructions
└── Bundled resources (optional)
    ├── scripts/     — executable code for deterministic/repetitive tasks
    ├── references/  — docs loaded into context as needed
    └── assets/      — files used in output (templates, icons, fonts)

渐进式披露——三级加载系统

技能分三个层级加载上下文，因此模型只需处理所需的内容：

1. 元数据（name + description）——始终位于上下文中（约 100 个单词）。这是触发条件，应在此投入精力。
2. SKILL.md 正文——每当技能触发时加载（理想情况下少于 500 行）。
3. 捆绑资源——仅在需要时加载（数量不限；脚本无需读入上下文即可执行）。

字数仅为近似值——如果技能确实需要，可以写得更长。

关键模式：

• 保持 SKILL.md 少于 500 行。若接近上限，请增加一层层级结构，并明确指出模型接下来应查阅的内容。
• 在 SKILL.md 中明确引用参考文件，并说明何时阅读各文件——切勿让 SKILL.md 一开始就加载所有文件（参见第 4 步）。
• 对于较大的参考文件（超过 300 行），请在顶部添加目录。

领域组织

当技能涵盖多个领域/框架时，按变体组织，并让模型只阅读相关文件——这与跨平台 Confluent 技能（references/confluent-cloud.md、references/warpstream.md 等）采用相同模式：

cloud-deploy/
├── SKILL.md          # workflow + variant selection
└── references/
    ├── aws.md
    ├── gcp.md
    └── azure.md

避免意外原则

用户阅读技能描述后，技能的行为不得让用户感到意外。技能绝不能包含恶意软件、漏洞利用代码或任何可能危及系统安全的内容，也不得助长未经授权的访问、数据外泄或其他恶意活动。无害的创意设定（例如“扮演 X”）没有问题，但具有误导性或恶意的技能不可以。

---

第 1 步：了解范围和平台定位

技能范围

了解用户想要构建什么。技能可以专注于用例，也可以专注于特定组件，两种方式都有效。

专注于用例 — 解决特定的端到端问题：

• “使用 Flink SQL 实时丰富客户订单”
• “使用 Debezium 和 Tableflow 构建从 PostgreSQL 到 Iceberg 的 CDC 流水线”
• “使用 Kafka Streams 实时检测异常”

专注于特定组件 — 涵盖 Confluent 组件的功能：

• “Flink SQL 技能”
• “Kafka 连接器技能”
• “Schema Registry 技能”
• “Tableflow 技能”

如果用户的请求不明确，请询问：

“这项技能应该涵盖什么？它可以针对特定用例（例如‘从 Postgres 到 Iceberg 的 CDC 流水线’），也可以专注于某个组件（例如‘Flink SQL 技能’）。你设想的范围是什么？”

平台定位

了解范围后，确定平台定位：

“这项技能应支持哪些平台？

1. 所有受支持的平台（Confluent Cloud、Confluent Platform、Apache Kafka、WarpStream）
2. 仅限特定平台（例如，仅限 Confluent Cloud 或仅限 WarpStream）”

如果针对特定平台：

• 平台名称必须出现在技能的 name 字段中（例如，confluent-cloud-cdc-tableflow、warpstream-producer-optimization）
• 技能描述应明确说明目标平台
• 测试必须针对该特定平台运行

如果跨平台：

• 技能名称不应包含平台前缀
• 生成的 SKILL.md 必须包含平台检测步骤，询问用户的目标平台
• 平台特有的细节应放在 references/<platform>.md 文件中：
  • references/confluent-cloud.md — Cloud 特有的配置、API 密钥和托管服务
  • references/confluent-platform.md — 自行管理环境的特有内容和安全配置
  • references/apache-kafka.md — OSS 默认设置，不含托管功能
  • references/warpstream.md — 对象存储架构和配置覆盖项
• SKILL.md 应有条件地指向这些文件：“如果用户的目标平台是 WarpStream，则阅读 references/warpstream.md 并应用其中的覆盖项”
• 参考模式请参阅 agent-skills 中的 PR #27

了解范围和平台后，继续进行访谈。

---

步骤 2-3：捕获意图并编写技能

按照基础技能创建流程，捕获用户意图并编写技能草稿。询问：

1. 此技能应帮助用户完成什么？

2. 什么情况下应触发？（用户可能会说的短语）

3. 涉及哪些组件？（主题、架构、Flink、连接器等）

4. **此技能应如何与 Confluent 交互？**询问用户：

   "此技能应如何与 Confluent 交互？

   1. MCP 工具（通过 Confluent MCP 服务——发现、管理和操作 Confluent 资源）
   2. CLI（Confluent CLI——confluent kafka topic create、confluent flink statement create 等）
   3. REST API（Confluent Cloud API、Schema Registry API、Kafka REST Proxy）
   4. 客户端库（confluent-kafka-python、Java 客户端等）
   5. 组合方式（例如，使用 MCP 进行发现 + 使用 CLI 进行资源配置 + 使用客户端库处理数据）"

   根据用户的回答，创建的技能必须包含：

   • 所需的具体工具/CLI/API 调用及其执行顺序
   • 任何前置设置（MCP 服务配置、CLI 登录、创建 API 密钥）
   • 每种交互方式的错误处理

5. 预期输出是什么？（可运行的管道、示例数据、SQL 语句等）

编写技能时：

• 添加指向相关 Confluent 文档的链接（docs.confluent.io）
• 引用具体的 API 端点、CLI 命令或 MCP 工具名称
• 清楚说明工具/CLI/API 调用的执行顺序（分步列出）
• 对常见操作使用随附脚本（参见 scripts/ 目录）
• 解释 Confluent 最佳实践背后的原因
• 链接到外部文档，让技能自动保持最新。若 docs.confluent.io 页面存在 .md 版本，优先使用该版本而非 .html（docs.confluent.io 会在相同路径提供简洁的 Markdown，便于 Agent 阅读），并在发布前确认每个链接都能正常访问：

  For Flink SQL syntax, see [Confluent Flink SQL Reference](https://docs.confluent.io/cloud/current/flink/reference/overview.md)
  

执行前规划要求

创建的技能 MUST 包含执行前规划步骤。技能执行任何操作（创建主题、注册架构、部署 Flink 语句等）之前，必须：

1. 向用户展示包含所有操作的编号计划
2. 列出将调用哪些工具/CLI 命令/API，以及调用顺序
3. 等待用户明确确认后再继续
4. 未经批准，绝不执行会修改资源的操作

计划格式及交互方式示例，请参阅 plan-template.md。

---

步骤 4：验证规范符合性

在继续测试之前，根据 Agent Skills 规范验证创建的技能。

Frontmatter 验证

• name：必填，最多 64 个字符，只能包含小写字母、数字和连字符，不得出现连续连字符，且必须与父目录名称一致
• description：必填，最多 1024 个字符，必须同时说明技能的功能和适用时机
• 如果特定于平台：平台名称必须出现在 name 中（例如，confluent-cloud-cdc-tableflow）
• metadata：必须包含 author: confluent、version（语义化版本字符串）和 last_updated（YYYY-MM-DD）
• compatibility：如果技能需要特定软件包、CLI 工具或环境访问权限，建议填写

触发重叠检查

在最终确定描述之前，检查它是否会与仓库中现有技能意外重叠：

1. 扫描目标仓库中的 skills/*/SKILL.md，提取每个技能的 description: 字段
2. 将新技能描述中的关键词与现有技能进行比较
3. 如果存在显著重叠（≥2 个非通用关键词相同），请确认边界是否明确——用户的提示应能明确地只触发一个技能，而非两个
4. 如果边界不明确，请调整新技能的描述以明确区分范围，或在必要时明确列出排除项

结构验证

• SKILL.md 正文不超过 500 行——将详细内容移至 references/
• 不要将参考文件内容内联到 SKILL.md 中——参考内容必须按需延迟加载。内联参考内容会让每次激活都付出完整的上下文成本。
• 参考文件应与 SKILL.md 相隔一层（不得形成嵌套引用链）
• 超过 300 行的大型参考文件应在顶部包含目录
• 如果 SKILL.md 超过约 200 行且涵盖多个工作流，应在顶部附近加入模式检测表

必需的文件

• evals/evals.json——每个主要用例至少包含一个测试用例，并遵循以下格式：

  {
    "skill_name": "skill-name",
    "evals": [
      {
        "id": 1,
        "prompt": "realistic user prompt (≥40 chars, specific context, not abstract)",
        "expected_output": "description of success",
        "files": [],
        "assertions": [
          "Specific, verifiable check with a path, identifier, or NOT clause",
          "Another assertion encoding hard-won correctness"
        ]
      }
    ]
  }
  

  评测质量规则（由 confluent-skill-reviewer 强制执行）：
  • 使用 assertions（字符串数组）——这是此代码库的标准。不要使用 expectations，也不要使用对象形式的条目（[{id, type, description, …}]）；check_eval_schema.py 会将两者标记为阻断问题
  • 每个评测都必须包含 files 字段（没有测试夹具时使用空数组 []）
  • 提示词必须至少包含 40 个字符，并且读起来像真实用户的消息（包括具体的数据形态和命名环境，不能只有“给我构建一个 X”）
  • 断言必须具体——包含文件路径、类名、配置键、NOT 子句或带引号的 CLI 标志。诸如“代码编写得很好”之类的模糊断言会被标记
  • 仅使用合成数据——提示词、测试夹具和样例记录中不得包含真实客户的个人身份信息或秘密（SSN、通过 Luhn 算法验证的银行卡号、AWS 密钥、私钥、真实邮箱/电话号码）。请使用 example.com/.example 域名、555-01xx 电话号码和明显的占位符。参见个人身份信息和合成数据；confluent-skill-reviewer 会阻止违规内容
• 如果技能需要凭据，则提供 .env.template 或 credentials.yaml.template
• 如果技能支持跨平台，则提供 references/<platform>.md 文件（每个受支持的平台一个）

运行验证

# If skills-ref is available
skills-ref validate ./created-skill

**如果验证失败：**修复问题并重新验证。技能通过验证前，不要继续测试。

---

第 5 步：设置测试环境和凭据

选择测试环境

询问用户要针对哪个环境进行测试：

“要针对哪个 Confluent 环境进行测试？

1. Local Confluent（Docker，用于开发）
2. Confluent Platform（本地自管理）
3. Confluent Cloud（托管式 SaaS）
4. WarpStream（兼容 Kafka，由对象存储提供支持）

注意：如果此技能针对特定平台，我们必须在该平台上进行测试。”

**重要规则：**如果技能是为特定平台设计的，测试就必须在该平台上运行。

安全性：凭据文件处理

凭据可能存放在 .env 文件或 YAML 文件（credentials.yaml）中——以下规则适用于该技能使用的任一格式。

重要：绝不要读取、显示或记录凭据文件（.env 或 credentials.yaml）及其内容。 完整安全策略请参阅 confluent-best-practices.md。关键规则：

• 绝不要对 .env 或 credentials.yaml 使用 cat、Read、head、grep 等命令
• 对于 .env：按名称引用变量（例如 $BOOTSTRAP_SERVERS），绝不读取其值；使用 test -n "$CLUSTER_API_KEY" 进行验证
• 对于 credentials.yaml：在使用它的脚本或进程内部加载（例如，在运行时使用 YAML 解析器），绝不将解析后的值回显到日志或标准输出
• 将凭据文件添加到 .gitignore（.env、credentials.yaml）
• 生成的技能必须包含相同的防护规则

重要：仅使用合成数据——绝不要提交真实客户 PII 或机密信息。 评估提示词、固定数据、示例记录和参考示例必须使用虚构数据（example.com 邮箱、555-01xx 电话号码、明显虚假的 ID）。绝不要嵌入真实 SSN、通过 Luhn 校验的支付卡号、AWS 密钥或私钥材料。请参阅 PII 和合成数据。打包前，可使用 python3 skills/confluent-skill-reviewer/scripts/scan_pii.py <skill-path> 检查技能目录。

设置凭据

有关特定环境的凭据模板，请参阅 凭据参考——模板提供 .env 和 YAML（credentials.yaml）两种形式，涵盖 Kafka、Schema Registry、Flink、Tableflow 和 Connectors。选择适合该技能的格式；YAML 适用于嵌套或结构化配置，.env 适用于扁平的 Shell 风格变量。

---

步骤 6：创建测试用例并运行端到端测试

测试用例结构

每个测试用例都应端到端地覆盖完整用例。有关 CRUD 操作计划格式，请参阅 plan-template.md。展示计划，并在创建任何资源前等待用户确认。

Schema Registry 默认设置

• 默认格式：JSON，标头中包含 Schema GUID（JSON_SR）
• 替代格式：Avro、Protobuf
• 询问用户是否要将 JSON_SR 更改为其他格式：“默认将使用带 Schema Registry 的 JSON 序列化（标头中包含 Schema GUID）。您想改用 Avro 还是 Protobuf？”
• 有关标头中 Schema GUID 的配置详情，请参阅 schema-guid-header.md

计算池检查（适用于 Flink 技能）

运行 Flink 操作前，确认有可用的计算池：

python scripts/check_compute_pool.py --pool-id $FLINK_COMPUTE_POOL_ID

如果没有可用的计算池或容量已满，请通知用户并跳过 Flink 测试。

运行测试

使用基础 skill-creator 工作流运行测试（启动有技能与无技能的子 Agent），但针对 Confluent 进行调整：

使用技能运行：

Execute this task:
- Skill path: <path-to-skill>
- Task: <eval prompt>
- Environment: .env file at <workspace>/.env
- Save outputs to: <workspace>/iteration-<N>/eval-<ID>/with_skill/outputs/
- Outputs to save:
  - Consumed data from output topic
  - SQL statements executed (if applicable)
  - Schema definitions
  - Verification logs

**基线运行：**使用相同提示词、不使用技能，并保持环境相同。

捆绑脚本

对于常见操作，请使用 scripts/ 中的捆绑脚本：

• check_compute_pool.py — 验证 Flink 计算池是否可用
• produce_data.py — 向主题生成 JSON_SR 编码的数据
• consume_and_verify.py — 消费数据并根据预期进行验证
• register_schema.py — 向 Schema Registry 注册架构
• cleanup_resources.py — 清理主题、架构和 Flink 语句（仅限本地环境）

如果脚本执行被拒绝（例如，无头／非交互式会话没有审批机制），请改用以下手动步骤，不要默默停滞：

• check_compute_pool.py → 使用 confluent flink compute-pool CLI 或 Cloud Console 手动检查计算池状态
• produce_data.py → 使用 confluent kafka topic produce CLI 手动生成 JSON_SR 编码的记录，或使用文中描述的一次性生产者代码片段
• consume_and_verify.py → 使用 confluent kafka topic consume 手动消费，并手动将数量／内容与预期进行比较
• register_schema.py → 使用 confluent schema-registry schema create 或 Schema Registry UI 手动注册架构
• cleanup_resources.py → 使用相应的 confluent CLI 删除命令手动删除主题／架构／语句（Cloud），如果环境是临时的，也可以跳过清理（仅限本地环境）

打包时，请将相关脚本包含在所创建技能的 scripts/ 目录中；如果所创建的技能自带脚本，也必须要求其记录相同的回退方案。

---

步骤 7-8：评估与迭代

遵循基础的 skill-creator 评估工作流：

1. 在测试运行期间起草断言（如适用）
2. 在运行完成时记录计时数据
3. 为每次运行评分
4. 汇总基准测试结果
5. 启动评估查看器
6. 与用户一起审阅反馈
7. 根据反馈改进技能

Confluent 专属断言——良好的断言应检查：

• 生成／消费的消息数量是否正确
• 架构兼容性（例如，向前／向后兼容）
• Flink 语句语法是否有效
• 输出数据是否包含预期的增强信息
• 是否没有数据丢失或重复
• 创建的主题是否采用正确的配置

---

步骤 9-10：优化描述并打包

描述优化

技能运行良好后，优化触发描述。重点关注：

• 特定用例关键词（数据增强、CDC、Tableflow 等）
• Confluent 产品名称（Flink、Kafka、Schema Registry 等）
• 常见用户表达（“设置管道”、“从……流式传输更改”等）
• 如果针对特定平台：在触发词中包含平台名称

打包

将以下内容与技能打包：

• SKILL.md，包含外部文档链接（少于 500 行），并包括：
  • 所有由 Confluent 创建的技能都应包含 metadata.author: confluent
  • metadata.version（semver 字符串，例如 "1.0"）
  • metadata.last_updated（YYYY-MM-DD）
  • 如果技能有环境或软件包要求，则包含 compatibility 字段
• evals/evals.json — 每个主要用例至少包含一个测试用例
• references/ — 特定平台的文件（如跨平台）以及任何 Tier 3 参考内容
• scripts/ — 相关的捆绑脚本
• 用于凭据的 .env.template 或 credentials.yaml.template

**验证关卡：**交付前，最后运行一次 skills-ref validate ./created-skill。验证失败的技能不得交付。

PR 提交清单

将技能添加到 Confluent 技能仓库（例如 confluentinc/agent-skills）时，请确保：

1. 更新 README — 在 README.md 的仓库技能表中添加新技能
2. 评估通过率达到 90%+ — 运行评估，并在 PR 描述中注明分数
3. SME 审核者 — 为该技能领域确定并指派一位主题专家
4. DTX/DevRel 审核者 — 指派一位 DTX 或 Developer Advocates 团队的审核者

这些是 PR 模板中的必需关卡，confluent-skill-reviewer 会对其进行检查。

---

参考文件

• credential-templates.md — 特定环境的凭据模板（.env 和 YAML）
• schema-guid-header.md — Header 配置中的 Schema GUID
• plan-template.md — CRUD 操作计划格式
• confluent-best-practices.md — 安全、错误处理、文档和测试指南
• example-enrichment.md — 完整的 enrichment 技能演练
• skill-creator 基础文档，介绍通用模式