developing-kafka-python-client

在明确询问并收到问题 #1（现有应用还是从零开始）、#2（目标环境）和 #3（生产者、消费者还是两者）的答案之前，不得生成任何代码、搭建任何项目或修改任何文件。如果用户的提示只回答了部分问题，仍须确认你的理解，然后才能生成。 无论提示看起来多么具体，此规则都适用。

首先宣布：“Using the Confluent Kafka Python Client skill to guide this project.”

Confluent Kafka Python 客户端创建

为使用 confluent-kafka-python 向 Kafka 生产消息和／或从 Kafka 消费消息，生成一个可用于生产环境的 Python 项目。支持三种目标环境：Confluent Cloud（托管）、Local Docker（开源 Kafka）和 WarpStream（兼容 Kafka、以对象存储为后端），以及两种生产者风格：AsyncIO（非阻塞）和 Synchronous（阻塞）。生成的代码遵循 Confluent 的最佳实践。

步骤 1：收集需求

生成任何代码之前，请先逐一询问以下问题。如果用户已在提示中明确回答某个问题，则跳过该问题——不要为了走形式而重复提问。例如，“在 Confluent Cloud 上构建一个生产者和消费者，并使用异步生产者”已回答 #2、#3 和 #4；只剩下 #1、#5、#6、#7 和 #8。

强制确认关卡——不得跳过，即使用户已回答所有问题也不例外。 写入任何文件之前，你必须发送一条消息，其中：

1. 用简短的项目符号列表概述你提取到的答案（例如：“目标：Confluent Cloud · 组件：生产者 + 消费者 · 生产者风格：异步 · 从零开始：是”）。
2. 在消息中提出所有尚未回答的问题。
3. 明确请用户确认或更正，然后再继续。

然后停止并等待用户回复。不要在发送概述的同一轮中生成文件；即使提示已充分说明需求，也不要假定用户同意立即生成——此概述用于发现对提示的误解，即使问题 #1–#8 都已预先回答，也必须执行。只有当用户已在本次对话中确认过该概述时，才能跳过此关卡。

不要对 #1、#2 或 #3 作出默认假设——如果提示没有回答其中任何一项，就必须询问。

1. 你是在现有应用中添加 Kafka，还是从头开始？
   • 如果用户有现成的 Python 代码（提到现有项目、有 main.py、使用 Flask/FastAPI/Django 等），不要搭建新项目。应当：（a）找出他们现有的生产者或发送数据的代码；（b）询问他们是否已在 Schema Registry 中注册架构；（c）按照参考文件中的模式，将 Schema Registry 集成到他们现有的代码中。只生成他们缺少的文件（例如 common.py、schemas/value.schema.json），并直接修改他们现有的代码。
   • 如果用户已经在不使用 Schema Registry 的情况下向 Kafka 生产数据（无架构），请帮助他们迁移：（1）根据现有消息结构生成 JSON Schema；（2）注册该架构；（3）将原始的 producer.produce() 调用替换为使用序列化器的调用。不要丢弃他们现有的代码。
   • 如果是从头开始，请按下面的完整脚手架流程进行。
2. 目标环境？——Confluent Cloud、本地 Kafka（Docker）还是 WarpStream。始终询问这一点，即使用户没有提到。如果用户提到“open source”“local”“docker”“self-hosted”，或只是想在没有云账户的情况下试用 Kafka，请选择本地 Docker。如果用户提到“Confluent Cloud”“CC”，或已有云凭据，请选择Confluent Cloud。如果用户提到“WarpStream”，请选择WarpStream。如果用户确认没有偏好，默认选择 Confluent Cloud，但始终要先询问。
   • **如果选择 WarpStream：**阅读 references/warpstream-optimization.md，并应用该参考文件中的 librdkafka 覆盖设置。主要变更：禁用幂等性，大幅增大批次大小和在途请求数，设置较大的抓取大小，并在 client.id 中添加 ws_az=<az> 以启用感知可用区的路由。除非需要基于实体的排序，否则优先使用空消息键进行粘性分区。
3. 生产者、消费者，还是两者都需要？
4. 异步还是同步生产者？（仅在需要生产者时询问。）帮助用户选择：
   • AsyncIO Producer（AIOProducer）：适用于运行在事件循环下且不能阻塞的代码，例如 FastAPI/Starlette、aiohttp、Sanic 和 asyncio 工作进程。
   • 同步 Producer（Producer）：适用于脚本、批处理作业，以及用户能够自行控制线程/进程并直接调用 poll()/flush() 的高吞吐量管道。 如果用户提到异步框架（FastAPI、aiohttp、Sanic）或使用 asyncio，默认选择AsyncIO。如果用户提到脚本、批处理、ETL，或没有偏好，默认选择同步。
5. **你是否有想使用的现有架构？**如果有，请让用户粘贴架构或提供文件路径，然后将其用作 schemas/value.schema.json，而不是生成新架构。如果没有，请继续询问他们的数据字段。
6. 你要生产哪类数据？（仅当用户没有现有架构时询问。获取字段名称和类型，以便生成匹配的 JSON Schema 和示例数据。）
7. 主题名称？（默认值：demo-topic）
8. 消费者组 ID？（仅在需要消费者时询问；默认值：python-consumer-group）

不要询问 Schema Registry——始终将其包含在方案中。对于 Confluent Cloud 和本地 Docker，始终使用 JSON Schema。如果目标是 WarpStream，请询问他们使用哪种 Schema Registry 实现——WarpStream 内置的 Schema Registry 仅支持 Avro 和 Protobuf（GET /schemas/types 返回 ["AVRO","PROTOBUF"]），因此如果他们使用该注册表，请询问他们倾向于 Avro 还是 Protobuf（默认选择 Avro）。如果他们使用其他 SR（例如 Confluent Cloud Schema Registry），则可以使用 JSON Schema。

常见的 Agent 错误

想法	实际情况
“用户提到了 FastAPI，所以我知道它是异步的——跳过提问”	仍要确认。用户可能想在 FastAPI 旁边运行一个同步后台工作进程。
“我会使用更常见的 Avro”	此技能默认使用 JSON Schema。**例外：**WarpStream 内置的架构注册表仅支持 Avro 和 Protobuf——如果用户使用的是 WarpStream SR，默认使用 Avro。如果用户使用的是其他 SR（例如 Confluent Cloud SR），无论 Kafka 环境如何，都可以使用 JSON Schema。
“我会跳过 Schema Registry，让事情简单些”	Schema Registry 不可或缺。每个项目都必须包含它。
“为了方便，我会将 auto.register.schemas=True”	始终设为 False。显式注册是核心原则。
“我会在 produce() 中创建生产者——这样更简洁”	只创建一个生产者实例，在 main() 中创建并将其作为参数传递。必须始终如此。
“用户要同步的，所以消费者也应该是同步的”	消费者始终是异步的（AIOConsumer）。这是有意为之的设计决策。
“我会为 AIOProducer 添加 headers=，以便传递架构 ID”	AIOProducer.produce() 在处理 headers 时会抛出 NotImplementedError。只有同步生产者使用 headers。
“我会将 AsyncJSONSerializer 换成 AsyncAvroSerializer，并保持调用位置不变”	JSONSerializer 将 schema_str 作为第一个参数；AvroSerializer 将 schema_registry_client 作为第一个参数。跨格式按位置传参会抛出 TypeError: ... got multiple values for argument 'schema_registry_client'。始终将两者都作为关键字参数传递。
“我会在生产者配置中设置 message.max.bytes=64000000，在消费者配置中设置 fetch.max.bytes=50242880——它们彼此独立”	在 librdkafka 中并非如此。message.max.bytes 是客户端全局配置（不同于 Java 中按角色区分的 max.request.size），因此当生产者和消费者共用 get_kafka_config() 时，消费者也会继承该配置。librdkafka 会在创建消费者时强制要求 fetch.max.bytes >= message.max.bytes，否则抛出 KafkaError{_INVALID_ARG, ... "fetch.max.bytes must be >= message.max.bytes"}。请使用 references/warpstream-optimization.md 中的 WarpStream librdkafka 消费者配置值（fetch.max.bytes=67108864）——这些值经过设定，能够满足此约束。

步骤 1b：确认理解

收集完所有答案后，在生成任何代码之前，先给出确认摘要：

Before I generate the project, let me confirm:
- Project type: [Greenfield scaffold / Migration of existing code]
- Environment: [Confluent Cloud (SASL_SSL) / Local Docker (PLAINTEXT) / WarpStream]
- Schema format: [JSON Schema / Avro / Protobuf] (Avro or Protobuf if using WarpStream's built-in SR)
- Components: [Producer only / Consumer only / Both]
- Producer style: [AsyncIO (AIOProducer) / Synchronous (Producer)] (if applicable)
- Schema: [brief description of user's data fields]
- Topic: [topic name]
- Consumer group: [group ID] (if consumer)

Does this look right?

等待用户确认后再继续执行步骤 2。如果用户更正了任何内容，请更新你的理解并再次确认。

步骤 2：生成项目

决策流程图

digraph decisions {
  "Q1: Existing app?" -> "Migration path:\nmodify existing code" [label="yes"];
  "Q1: Existing app?" -> "Q2: Environment?" [label="no / greenfield"];
  "Q2: Environment?" -> "Cloud config\n(SASL_SSL)" [label="Confluent Cloud"];
  "Q2: Environment?" -> "Local Docker config\n(PLAINTEXT) + docker-compose.yml" [label="local / docker / OSS"];
  "Q2: Environment?" -> "WarpStream config\n(apply overrides from\nreferences/warpstream-optimization.md)" [label="WarpStream"];
  "Cloud config\n(SASL_SSL)" -> "Q3: Components?";
  "Local Docker config\n(PLAINTEXT) + docker-compose.yml" -> "Q3: Components?";
  "WarpStream config\n(apply overrides from\nreferences/warpstream-optimization.md)" -> "Which SR?\n(WarpStream SR → Avro/Protobuf;\nother SR → JSON Schema)";
  "Which SR?\n(WarpStream SR → Avro/Protobuf;\nother SR → JSON Schema)" -> "Q3: Components?";
  "Q3: Components?" -> "Q4: Async or sync?" [label="producer requested"];
  "Q3: Components?" -> "Generate consumer\n(always async AIOConsumer)" [label="consumer only"];
  "Q4: Async or sync?" -> "AIOProducer path\nAsyncJSONSerializer\n(no headers support)" [label="async / event-loop"];
  "Q4: Async or sync?" -> "Producer path\nJSONSerializer\n(header-based schema ID)" [label="sync / batch / ETL"];
}

在用户选择的目录中创建以下文件结构：

<project-dir>/
├── producer.py          # (if requested)
├── consumer.py          # (if requested)
├── common.py            # shared config loading + verification helpers
├── schemas/
│   └── value.schema.json # JSON Schema (or value.avsc / value.proto when using WarpStream SR)
├── tests/
│   └── test_project.py  # unit tests (always generated)
├── .env.example         # template for credentials
├── requirements.txt
├── docker-compose.yml   # (local Docker path only)

安全

绝不读取、打开或显示 .env 文件。它们包含 API 密钥和机密信息。只生成包含占位值的 .env.example。如果用户要求排查连接问题，请让他们自行验证 .env 中的值——不要读取该文件。

核心原则

这些原则很重要，因为它们能避免 Kafka Python 客户端最常见的生产问题：

1. 复用生产者实例。 每条消息都创建新的生产者代价高昂——每个实例都会建立新的 TCP 连接、执行 SASL 握手并获取元数据。创建一个生产者并将其用于所有消息。生产函数应将生产者作为参数接收，而不是自行实例化。

2. 始终将 Schema Registry 与 JSON Schema 配合使用。 Schema Registry 会在生产者和消费者之间强制执行契约。没有它，架构更改会悄无声息地导致下游消费者出错。此技能默认使用 JSON Schema。Schema Registry 支持 Avro、Protobuf 和 JSON Schema——选择 JSON Schema 的原因是：（1）Python 原生支持 JSON，无需代码生成步骤；（2）confluent-kafka-python 开箱即提供 JSONSerializer/JSONDeserializer；（3）对于已经使用 JSON/dict 数据的 Python 开发者来说，这是最容易上手的格式。 WarpStream Schema Registry 例外： WarpStream 内置的 Schema Registry 仅支持 Avro 和 Protobuf——其 GET /schemas/types 端点返回 ["AVRO","PROTOBUF"]。使用 WarpStream 的 SR 时，不支持 JSON Schema。如果用户运行的是 WarpStream，并使用其内置 SR，默认使用 Avro（或者用户更喜欢的话，也可以使用 Protobuf）。使用 confluent_kafka.schema_registry.avro 中的 AvroSerializer/AvroDeserializer（或 confluent_kafka.schema_registry.protobuf 中的 ProtobufSerializer/ProtobufDeserializer）。对于异步生产者，使用 confluent_kafka.schema_registry._async.avro 中的 AsyncAvroSerializer（或 confluent_kafka.schema_registry._async.protobuf 中的 AsyncProtobufSerializer）。生成 Avro schema 文件 schemas/value.avsc（Protobuf 则生成 schemas/value.proto），而不是 schemas/value.schema.json——以 references/value.avsc 为起点，并根据用户的领域进行调整。在生成 Avro 路径的生产者/消费者代码时，请从 references/producer_avro.py（异步）、references/producer_avro_sync.py（同步）和 references/consumer_avro.py（异步）复制——不要使用 JSON 参考文件——因为构造函数签名不同（见下文的 Avro 构造函数警告）。如果用户运行的是 WarpStream，但连接的是其他 SR（例如 Confluent Cloud Schema Registry），则可以正常使用 JSON Schema——采用默认路径。

   在创建序列化器之前，先将 schema 注册为单独的显式步骤。 使用专门的 register_schema() 函数调用 sr_client.register_schema()，并让错误（身份验证失败、网络错误、权限拒绝）立即向上传播——绝不要用裸 try/except 包装注册操作。然后将序列化器配置为 auto.register.schemas=False 和 use.latest.version=True。这样可以确保序列化器不会静默自动注册，并符合生产实践：由 CI/CD 注册 schema，而不是由应用启动时注册。

   根据选定的生产者风格和 schema 格式，使用相应的序列化器：

   • JSON Schema（默认）： 异步场景使用 confluent_kafka.schema_registry._async.json_schema 中的 AsyncJSONSerializer / AsyncJSONDeserializer；同步场景使用 confluent_kafka.schema_registry.json_schema 中的 JSONSerializer / JSONDeserializer。
   • Avro（使用 WarpStream SR 时的默认选项）： 异步场景使用 confluent_kafka.schema_registry._async.avro 中的 AsyncAvroSerializer / AsyncAvroDeserializer；同步场景使用 confluent_kafka.schema_registry.avro 中的 AvroSerializer / AvroDeserializer。
   • Protobuf（使用 WarpStream SR 时的替代选项）： 异步场景使用 confluent_kafka.schema_registry._async.protobuf 中的 AsyncProtobufSerializer / AsyncProtobufDeserializer；同步场景使用 confluent_kafka.schema_registry.protobuf 中的 ProtobufSerializer / ProtobufDeserializer。

3. 选择合适的生产者风格。 confluent-kafka-python 库提供两种生产者 API：

   • AsyncIO Producer（来自 confluent_kafka.aio 的 AIOProducer）：非阻塞，可与 asyncio 事件循环集成。与 confluent_kafka.schema_registry._async.json_schema 中的 AsyncJSONSerializer 以及 AsyncSchemaRegistryClient 配合使用。最适合已运行事件循环的应用（FastAPI、aiohttp、Sanic、asyncio worker）。
   • 同步 Producer（来自 confluent_kafka 的 Producer）：阻塞调用，使用投递回调。与 confluent_kafka.schema_registry.json_schema 中的 JSONSerializer 以及 SchemaRegistryClient 配合使用。最适合脚本、批处理作业，以及吞吐量最高的管道；在这类场景中，用户可以控制线程/进程并直接调用 poll()/flush()。 务必询问用户哪种风格适合其使用场景。消费者始终使用 AIOConsumer（异步）——长时间运行的轮询循环受益于非阻塞 I/O，而混用同步/异步消费者风格只会增加复杂性，收益却很少。

4. 平滑关闭。 异步生产者必须在退出前调用 flush() 和 close()（两者都要等待完成）。同步生产者必须在退出前调用 flush()，否则缓冲的消息会丢失。消费者必须先调用 unsubscribe()，再调用 close()，以便干净地离开消费者组（避免不必要的重新均衡）。使用 try/finally 代码块，并处理 KeyboardInterrupt / 信号。

5. 支持 Confluent Cloud、本地 Docker 和 WarpStream。 使用 Confluent Cloud 时，配置 SASL_SSL 和 PLAIN 机制，并从 .env 加载 API 密钥。使用本地 Docker 时，使用 PLAINTEXT，且不进行身份验证。使用 WarpStream 时，根据用户的 WarpStream 部署情况使用 SASL_SSL 或 PLAINTEXT，并应用 references/warpstream-optimization.md 中的 librdkafka 覆盖设置（大批次、禁用幂等性、大型拉取、支持区域感知的 client.id）。环境变量 KAFKA_ENV（cloud、local 或 warpstream）用于控制所使用的路径。通过 .env 从环境变量加载所有设置。

6. 运行前验证连接。 使用 AdminClient.list_topics() 验证代理可连接且主题存在，然后再进行生产或消费。通过 HTTP 健康检查验证 Schema Registry 的连接。

7. 始终为领域事件设置消息键。 对于任何代表实体或事件流的消息（订单事件、用户操作、设备遥测、交易），向 producer.produce() 传入 key=<entity_id>.encode("utf-8")。Kafka 根据键对消息分区，因此键相同的消息会进入同一分区并保持顺序——这对于 OrderCreated → OrderUpdated → OrderCancelled 等事件流至关重要，消费者必须按顺序看到这些事件。每个参考文件中的 produce() 辅助函数都接受 key_field 参数，用于指定作为键的字段（例如，key_field="order_id"、key_field="transaction_id"）。询问用户哪个字段用于标识实体，并将其传给 produce()。只有在用户明确表示不需要考虑顺序时（例如无状态指标，分区无关紧要），才将 key_field=None。

   WarpStream 例外： 在 WarpStream 上，空键可启用粘性分区，从而构建更大的批次，并显著提升吞吐量、降低成本。如果用户的用例不需要按实体排序（例如彼此独立的遥测读数、无状态指标、日志），建议使用 key_field=None，并说明其吞吐量优势。如果需要按实体排序（例如同一订单的 OrderCreated → OrderUpdated → OrderCancelled），仍应设置消息键——正确性优先于批处理效率。询问用户事件是否需要按键排序，以便作出判断。

common.py

此模块负责加载配置并验证连接。使用 references/common.py 作为模板。

producer.py 模式（AsyncIO）

当用户选择 AsyncIO producer 时，请使用 references/producer.py 作为模板。

要点：

• produce() 接受一个 producer 实例作为参数——它不会创建实例
• 在 main() 中只创建一次 producer，并可将其传递给多个 produce() 调用
• 调用异步序列化器（AsyncJSONSerializer）处理消息时，必须使用 await
• AIOProducer.produce() 是异步的，会返回一个 asyncio.Future。必须先对该方法使用 await 以获取 Future，再对 Future 使用 await 以获取已投递的 Message：future = await producer.produce(...); result = await future
• AIOProducer.flush() 和 close() 是协程——必须在 finally 块中使用 await
• 信号处理程序会设置一个关闭事件，以实现平稳终止
• Schema 注册和序列化器创建是分开的步骤。register_schema() 会显式注册 schema 并返回 schema ID——错误会立即向上传播。create_json_serializer()（或使用 Avro/Protobuf 时的 create_avro_serializer() / create_protobuf_serializer()）会通过 conf={'auto.register.schemas': False, 'use.latest.version': True} 创建序列化器。
• 不同格式的构造函数参数顺序各不相同——始终将 schema_registry_client 和 schema_str 作为关键字参数传递，绝不要按位置传递。 confluent-kafka-python 中的 JSON、Avro 和 Protobuf 序列化器/反序列化器类并不共用相同的位置参数签名：JSONSerializer/JSONDeserializer 将 schema_str 作为第一个参数，而 AvroSerializer/AvroDeserializer 和 ProtobufSerializer/ProtobufDeserializer 将 schema_registry_client 作为第一个参数。在不同格式中混用位置参数和关键字参数会导致 TypeError: ... got multiple values for argument 'schema_registry_client'。请在所有地方使用关键字参数，使模式保持一致：
  • await AsyncJSONSerializer(schema_str=schema_str, schema_registry_client=sr_client, conf=conf)
  • await AsyncAvroSerializer(schema_str=schema_str, schema_registry_client=sr_client, conf=conf)
  • await AsyncProtobufSerializer(msg_type=MsgType, schema_registry_client=sr_client, conf=conf)（Protobuf 接受生成的消息类，而不是 schema 字符串） 规范模板请参阅 references/producer.py（JSON）和 references/producer_avro.py（Avro）。
• AIOProducer 批处理模式不支持 Headers。 不要向 AIOProducer.produce() 传递 headers=——这会引发 NotImplementedError。Schema 标识由序列化器的 wire format 前缀自动处理（适用于 JSON Schema、Avro 和 Protobuf 序列化器）。详情请参阅下方的“Schema ID in Headers vs Wire Format”

producer.py 模式（同步）

当用户选择同步生产者时，请使用 references/producer_sync.py 作为模板。

要点：

• produce() 接收一个生产者实例作为参数，绝不会自行创建生产者
• 在 main() 中创建一次生产者，并可将其传递给多次 produce() 调用
• 使用 confluent_kafka.schema_registry.json_schema 中的 JSONSerializer（同步）和 confluent_kafka.schema_registry 中的 SchemaRegistryClient。使用 Avro/Protobuf（例如，配合 WarpStream SR）时，使用 confluent_kafka.schema_registry.avro 中的 AvroSerializer，或 confluent_kafka.schema_registry.protobuf 中的 ProtobufSerializer
• Producer.produce() 是非阻塞的，会将消息加入队列。每次调用 produce 后都要调用 producer.poll(0)，以处理投递回调并防止内部队列填满
• 批次处理完成后调用 producer.flush()，阻塞至所有正在传输的消息均已投递
• 使用 delivery_callback(err, msg) 函数处理每条消息的投递报告
• 信号处理器设置一个标志，以便优雅终止
• 在 finally 代码块中调用 flush()，确保不会丢失缓冲的消息
• Schema 注册和序列化器创建是分开的步骤，与异步模式相同。register_schema() 会显式注册 Schema。create_json_serializer()（或使用 Avro/Protobuf 时的 create_avro_serializer() / create_protobuf_serializer()）会创建序列化器，并使用 conf={'auto.register.schemas': False, 'use.latest.version': True}。两者都会返回 Schema ID
• 同步版本同样只能使用关键字参数：调用 JSONSerializer(schema_str=..., schema_registry_client=..., conf=...)、AvroSerializer(schema_str=..., schema_registry_client=..., conf=...) 或 ProtobufSerializer(msg_type=..., schema_registry_client=..., conf=...)。不同格式的位置参数顺序并不一致，以位置参数传递会触发 TypeError: ... got multiple values for argument 'schema_registry_client'。模板请参阅 references/producer_sync.py（JSON）和 references/producer_avro_sync.py（Avro）
• 每条生成的消息都通过 Kafka 记录头（confluent.value.schemaId）传递 Schema ID，这是基于头部的 Schema 标识模式。这样 JSON 负载保持整洁，非 Confluent 消费者也能读取。详情请参阅下文“Schema ID 在头部与线格式中的区别”

consumer.py 模式

请使用 references/consumer.py 作为模板。

消费者要点：

• 基于信号实现优雅关停：先调用 unsubscribe()，再调用 close()，以便干净地离开消费者组
• 通过 Schema Registry 进行反序列化，使用 AsyncJSONDeserializer（使用 Avro/Protobuf 时则使用 AsyncAvroDeserializer / AsyncProtobufDeserializer），不回退到原始解析；Schema Registry 是必需的。构造反序列化器时只能使用关键字参数，例如：await AsyncAvroDeserializer(schema_str=schema_str, schema_registry_client=sr_client)。不同格式的位置参数签名不同（Avro/Protobuf 将 schema_registry_client 放在首位；JSON 将 schema_str 放在首位），混用位置参数和关键字参数会触发 TypeError: ... got multiple values for argument 'schema_registry_client'。模板请参阅 references/consumer.py（JSON）和 references/consumer_avro.py（Avro）
• 持续轮询，直到收到关停信号

请求头中的 Schema ID 与线格式

**同步生产者：**参考代码会将 Schema ID 作为 Kafka 记录请求头（confluent.value.schemaId）传入每条消息。这能保持 JSON 负载简洁（不带魔数前缀），便于非 Confluent 消费者读取，也能使用 kcat 等工具进行调试。这是同步生产者的推荐做法。

异步生产者（AIOProducer）：AIOProducer 在批处理模式下不支持自定义请求头（传入 headers= 时，produce() 会引发 NotImplementedError）。Schema 标识依赖 JSON Schema 序列化器的线格式前缀（在负载前添加魔数字节 + Schema ID）。这是已知限制——不要尝试为异步生产的消息添加请求头。

**如何选择：**如果下游消费者都能识别 Confluent（使用 Schema Registry 反序列化器），两种方式都能透明地工作。如果下游消费者不使用 Confluent（普通 JSON 消费者），则更推荐使用基于请求头传递 Schema ID 的同步生产者，因为消息值仍是简洁的 JSON。面向混合消费者生态生成 README 时，请说明这一取舍。

schemas/

根据用户的数据领域，以所选 Schema Registry 适用的格式生成匹配的 Schema 文件。

**JSON Schema（默认）：**在 schemas/value.schema.json 生成 JSON Schema 文件。

例如，如果用户要生成金融交易数据：

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Transaction",
  "description": "A financial transaction event produced to Kafka.",
  "type": "object",
  "properties": {
    "transaction_id": {
      "type": "string",
      "description": "Unique identifier for this transaction."
    },
    "amount": {
      "type": "number",
      "description": "Transaction amount in the specified currency.",
      "default": 0
    },
    "currency": {
      "type": "string",
      "description": "ISO 4217 currency code.",
      "default": ""
    },
    "timestamp": {
      "type": "string",
      "format": "date-time",
      "description": "Time the transaction occurred, in ISO 8601 format."
    },
    "status": {
      "description": "Current state of the transaction.",
      "enum": ["pending", "completed", "failed", "refunded"],
      "default": "pending"
    },
    "metadata": {
      "oneOf": [{"type": "null"}, {"type": "object"}],
      "description": "Optional metadata associated with the transaction.",
      "default": null
    }
  },
  "required": ["transaction_id", "amount", "currency", "timestamp", "status"]
}

**Avro（使用 WarpStream SR 时）：**在 schemas/value.avsc 生成 Avro Schema 文件（Protobuf 则使用 schemas/value.proto）。以下是相同金融交易示例的 Avro 格式：

{
  "type": "record",
  "name": "Transaction",
  "namespace": "com.example.kafka",
  "doc": "A financial transaction event produced to Kafka.",
  "fields": [
    {"name": "transaction_id", "type": "string", "doc": "Unique identifier for this transaction."},
    {"name": "amount", "type": "double", "default": 0, "doc": "Transaction amount in the specified currency."},
    {"name": "currency", "type": "string", "default": "", "doc": "ISO 4217 currency code."},
    {"name": "timestamp", "type": "string", "doc": "Time the transaction occurred, in ISO 8601 format."},
    {"name": "status", "type": {"type": "enum", "name": "Status", "symbols": ["pending", "completed", "failed", "refunded"]}, "default": "pending", "doc": "Current state of the transaction."},
    {"name": "metadata", "type": ["null", "string"], "default": null, "doc": "Optional metadata associated with the transaction."}
  ]
}

Schema 生成规则

生成或调整符合用户领域的 Schema 时，请严格遵循 references/schema-generation-rules.md 中的规则。

多事件主题（高级）

当用户描述在单个主题上使用多种事件类型时，请遵循 references/multi-event-guide.md。只有在用户明确描述单个主题上存在多种事件类型时，才建议使用多事件联合 Schema。

docker-compose.yml（仅适用于本地 Docker 路径）

当用户选择本地 Docker 时，必须使用 references/docker-compose.yml 作为模板生成 docker-compose.yml。这会启动一个单节点 Kafka broker（使用 confluentinc/confluent-local）和一个 Confluent Schema Registry。用户只需运行 docker compose up -d，即可获得可用的 Kafka 环境。

重要：confluentinc/confluent-local 镜像使用 KRaft 模式，并内置以下监听器名称：PLAINTEXT（内部，端口 29092）、PLAINTEXT_HOST（外部，端口 9092）和 CONTROLLER（端口 29093）。不要发明自定义监听器名称，否则会与镜像的内部配置冲突，并导致启动循环。只能使用这些确切的监听器名称覆盖 KAFKA_ADVERTISED_LISTENERS 和 KAFKA_LISTENERS。内部 PLAINTEXT 监听器必须广播 kafka 主机名（而不是 localhost），以便 Schema Registry 能够从 Docker 网络内部访问 broker。

.env.example

根据目标环境生成适当的 .env.example：

Confluent Cloud：

KAFKA_ENV=cloud
BOOTSTRAP_SERVER=pkc-xxxxx.us-east-1.aws.confluent.cloud:9092
API_KEY=your-api-key
API_SECRET=your-api-secret
TOPIC=demo-topic
SCHEMA_REGISTRY_URL=https://psrc-xxxxx.us-east-2.aws.confluent.cloud
SR_API_KEY=your-sr-api-key
SR_API_SECRET=your-sr-api-secret
CLIENT_ID=python-client
GROUP_ID=python-consumer-group

本地 Docker：

KAFKA_ENV=local
BOOTSTRAP_SERVER=localhost:9092
TOPIC=demo-topic
SCHEMA_REGISTRY_URL=http://localhost:8081
CLIENT_ID=python-client
GROUP_ID=python-consumer-group

WarpStream：

KAFKA_ENV=warpstream
BOOTSTRAP_SERVER=your-warpstream-bootstrap-url:9092
TOPIC=demo-topic
SCHEMA_REGISTRY_URL=http://your-schema-registry:8081
CLIENT_ID=python-client,ws_az=us-east-1a
GROUP_ID=python-consumer-group
# If your WarpStream deployment requires SASL auth, uncomment:
# API_KEY=your-api-key
# API_SECRET=your-api-secret
# If using Confluent Cloud Schema Registry, uncomment:
# SR_API_KEY=your-sr-api-key
# SR_API_SECRET=your-sr-api-secret

requirements.txt

JSON Schema（默认）：

confluent-kafka[json,schema_registry]>=2.13.2
jsonschema
python-dotenv
requests>=2.25.0
httpx
authlib
cachetools
attrs
typing_extensions
pytest
pytest-asyncio

Avro（例如，使用 WarpStream SR 时）：

confluent-kafka[avro,schema_registry]>=2.13.2
python-dotenv
requests>=2.25.0
httpx
authlib
cachetools
attrs
typing_extensions
pytest
pytest-asyncio

Protobuf（例如，使用 WarpStream SR 时）：

confluent-kafka[protobuf,schema_registry]>=2.13.2
python-dotenv
requests>=2.25.0
httpx
authlib
cachetools
attrs
typing_extensions
pytest
pytest-asyncio

生成代码（producer.py、consumer.py、common.py）中任何位置导入的每个第三方包，都必须在 requirements.txt 中有对应条目。如果代码使用 from confluent_kafka import ...，那么 requirements.txt 中必须包含 confluent-kafka。如果代码使用 from dotenv import load_dotenv，那么必须列出 python-dotenv。这也包括不会自动安装的传递依赖——例如，异步 Schema Registry 客户端会在运行时导入 httpx 和 authlib，因此即使它们未声明为 confluent-kafka 的依赖，也必须明确列出。用户应能运行 pip install -r requirements.txt 并执行代码，且不会遇到任何 ModuleNotFoundError。

始终包含 pytest。如果项目使用异步生产者或消费者，则包含 pytest-asyncio。只有当生产者使用 Faker 生成示例数据时，才包含 Faker。

README.md

按照 references/readme-template.md 生成 README。根据实际生成的内容进行调整——如果只要求生成消费者，则省略生产者部分；对于 Confluent Cloud 项目，则省略 Docker 部分。

tests/test_project.py

始终生成单元测试。使用 references/test_project.py 作为模板。测试不得依赖正在运行的 Kafka 集群或 Schema Registry——模拟所有外部依赖，使测试能在 CI 和评测环境中通过。

测试应验证生成代码具备以下属性：

1. common.py：load_config() 返回所有必需键并使用正确的默认值。get_kafka_config() 在 KAFKA_ENV=cloud 时生成包含 SASL_SSL 和 PLAIN 的配置；在 KAFKA_ENV=local 时生成不含 SASL 的 PLAINTEXT 配置。模拟成功或失败时，verify_kafka_setup() 和 verify_schema_registry() 会返回正确的布尔值。

2. producer.py（如果已生成）：produce() 接受生产者实例和 schema_id 作为参数（绝不创建生产者）。生产者类（异步时为 AIOProducer，同步时为 Producer）在模块中只实例化一次。消息在生产前会经过序列化器处理。每条已生成的消息都将 Schema ID 包含在 Kafka 记录头 confluent.value.schemaId 中。对于同步生产者，验证生产后调用了 flush()。

3. consumer.py（如果已生成）：使用 Schema Registry 反序列化器（根据所选格式，使用 JSONDeserializer/AsyncJSONDeserializer、AvroDeserializer/AsyncAvroDeserializer 或 ProtobufDeserializer/AsyncProtobufDeserializer）——不得回退到原始的 json.loads。为实现平稳关闭，在调用 close() 前先调用 unsubscribe()。

4. Schema 文件：测试模板包含三个 Schema 类——TestJsonSchema、TestAvroSchema、TestProtobufSchema——每个类都由 pytest.mark.skipif 保护，并检查对应文件（value.schema.json、value.avsc、value.proto）是否存在。所选格式对应的类会运行；其他类会跳过。JSON Schema 检查：type: object、title、至少包含一个属性的 properties、所有属性都有描述，以及类似时间戳的字段具有 format: date-time。Avro 检查：type: record、name、至少包含一个字段的 fields、记录和每个字段都有 doc。Protobuf 检查：proto3 语法以及至少一个 message 定义。

5. 项目结构：requirements.txt 存在，且包含 confluent-kafka、python-dotenv 和 requests。.env.example 存在。

根据用户特定的 Schema 和数据领域调整测试——如果用户的字段包括 device_id 和 temperature，Schema 测试可以检查这些特定字段名称。

生成所有文件后，运行 pytest tests/ 验证测试是否通过。如果有测试失败，应修复生成的代码（而不是测试），直至测试通过。

第 3 步：指导用户

生成文件后，根据用户的目标环境向其提供说明。根据实际生成的内容调整说明：如果用户只请求了生产者，则省略消费者步骤；如果用户只请求了消费者，则省略生产者步骤。

Confluent Cloud：

1. 将 .env.example 复制为 .env，并填写 Confluent Cloud 凭据
2. 创建 virtualenv 并安装依赖项：pip install -r requirements.txt
3. 如果生成了生产者：首次运行生产者时，会通过 register_schema() 函数显式注册 schema（auto.register.schemas 设置为 False）。如果只生成了消费者：将 schemas/value.schema.json 的内容粘贴到 Confluent Cloud Console 中的 Schema Registry > Schemas 下，为主题的 value subject 手动注册 schema。
4. 运行生产者（如果已生成）：python producer.py
5. 运行消费者（如果已生成）：python consumer.py

提醒用户，他们可以在 Confluent Cloud Console 的集群和环境设置中找到 bootstrap server、API 密钥和 Schema Registry URL。

Local Docker：

1. 启动 Kafka 和 Schema Registry：docker compose up -d
2. 将 .env.example 复制为 .env（已预先填入 Local Docker 的默认值，无需修改）
3. 创建 virtualenv 并安装依赖项：pip install -r requirements.txt
4. 创建主题（如果禁用了自动创建）：docker compose exec kafka kafka-topics --create --topic demo-topic --bootstrap-server localhost:29092
5. 运行生产者（如果已生成）：python producer.py
6. 运行消费者（如果已生成）：python consumer.py
7. 完成后，停止容器：docker compose down（添加 -v 还会删除存储的数据）

WarpStream：

1. 将 .env.example 复制为 .env，并填写 WarpStream bootstrap server、Schema Registry URL 和（如适用）凭据。将 CLIENT_ID 设置为包含 ws_az=<availability-zone>，以启用区域感知路由（例如：python-client,ws_az=us-east-1a）
2. 创建 virtualenv 并安装依赖项：pip install -r requirements.txt
3. 如果主题不存在，则创建主题：kafka-topics.sh --create --topic demo-topic --bootstrap-server <warpstream-agent>:9092
4. 运行生产者（如果已生成）：python producer.py
5. 运行消费者（如果已生成）：python consumer.py

提醒用户，WarpStream 的生产延迟高于标准 Kafka（约为 250ms p50），这是预期行为。如果吞吐量低于预期，请确认已应用 references/warpstream-optimization.md 中针对 WarpStream 的配置覆盖项（尤其是 enable.idempotence=false 和较大的 batch/fetch 大小）。