最近我们在生产环境完成了一件让团队都比较兴奋的事：把线上正在跑的 Kafka 集群正式接入了 AI 推理服务。以前 Kafka 在我这儿就是个纯管道工具，负责把订单、日志、用户行为这些消息从上游搬到下游，业务逻辑大多是固定的规则和清洗。现在不一样了，消息流经 Kafka 后，会先被交给大模型做实时理解、打标签、分类，甚至是生成回复，然后再把结果送进下一个环节。听起来只是“给消息队列加了个 AI 中间层”，但真正落地的时候涉及到的细节远比想象中多。
这个方案适合谁参考？如果你正在做 AI Agent、智能客服、内容审核、实时舆情分析，或者是想把自己本地部署的大模型接到业务消息流上，这篇文章应该能帮你省不少弯路。我会把整个接入过程的架构思路、核心实现、参数取舍，以及我踩过的几个坑都拆开讲一遍，尽量让看完的人能直接照着做。
1. 内容整体设计与思路拆解
1.1 为什么是 Kafka 承载 AI 数据流
先说个很现实的场景：我们有个系统每天要处理上百万条用户反馈消息，之前的处理方式是定时任务批量捞数据，然后调用外部 AI 接口做一轮分析，结果写入数据库。这个方案有几个痛点，一是定时任务有延迟，无法做到“消息进来就被处理”；二是高峰期批量调用 AI 接口，经常把服务打满，超时和失败率都很难看；三是一旦 AI 服务升级或者换模型，历史数据想重新分析一遍会很麻烦，因为原始消息可能已经找不回来了。
Kafka 天然适合解决这些问题。它是基于发布订阅模式的消息队列，消息进来之后会持久化在磁盘上，消费者可以按照自己的速度拉取数据，这就实现了生产端和消费端的完全解耦。当业务高峰期涌入大量消息时，Kafka 充当一个巨大的缓冲池，AI 服务不会被打爆；当 AI 服务需要维护升级时可以随时停掉消费者，消息不会丢，等恢复后还能从上次的位置继续消费。
更关键的一点是消息回放能力。我们的业务有一个刚需：换了更强的模型之后，需要用历史数据重新跑一遍分析结果。Kafka 里消息默认保留 7 天，如果需要更长的回放窗口，可以调整保留策略。这意味着 AI 分析结果哪怕被写入了其他存储，只要原始消息还在 Kafka 里，换模型重新分析就是一个重启消费者的操作，不需要再找业务方要数据。
有人可能会说，直接用 HTTP 同步调用 AI 接口不就行了，为什么要绕一圈 Kafka？我之前也是这么干的，后来发现当调用量上来之后，同步调用对业务系统的侵入性太强。业务系统需要等待 AI 服务返回结果，AI 一慢整个业务流程就阻塞，超时重试的逻辑也让代码变得特别复杂。把请求扔进 Kafka，业务系统立刻得到响应，AI 服务异步处理，处理完再回调或者把结果写入结果队列，这才是生产环境该有的玩法。
1.2 技术选型：一边是消息管道，一边是 AI 服务
接入方案里有两个核心角色：消息管道（Kafka）和 AI 推理服务。我这次用的是本地部署的开源模型，而不是调用云端大模型 API，主要原因是业务数据涉及用户隐私，不能随便出内网。本地推理服务经过对比之后选了 Ollama，因为它对硬件要求相对友好，部署起来也简单，一条命令就能拉起来一个 OpenAI 兼容的 API 服务。如果你的场景对并发要求更高，预算也充足，可以考虑 vLLM，吞吐量确实强很多，但显存要求也更苛刻。
Kafka 这边用的是 3.0.0 版本，和热词里出现的 kafka_2.13-3.0.0 对应，这也是目前生产环境里比较稳的组合。客户端这边我们同时用了 Python 和 Java 两套，Python 负责灵活地做数据处理和模型调用，Java 那边对接集团内部的 Spring Boot 项目。这种混合架构实践中完全没问题，Kafka 的协议是多语言兼容的，只要保证消息格式统一就行。
这里补充一个选型心得。如果是新项目，我更推荐直接用 confluent-kafka-python 这个客户端库，它是基于 C 语言 librdkafka 封装的，性能和稳定性都比纯 Python 写的 kafka-python 要好不少。我们早期用过 kafka-python，消费速度上来之后经常出现连接被重置和 rebalance 问题，后来切换到 confluent-kafka 之后情况明显改善。如果是 Java 技术栈，spring-kafka 已经封装得很好了，配置一下就能用。
1.3 三种接入模式，针对不同场景选型
我们在设计接入方案时，发现“Kafka 接入 AI”这个需求其实可以拆成三种不同的模式，不同模式的实现差别很大。
第一种是流式实时推理模式。消息进入 Kafka 后被 AI 服务实时消费，立刻做推理，结果写回另一个 Kafka topic 或者直接入库。这种模式适合对实时性要求高的场景，比如在线客服的实时话术推荐、风控系统的实时拦截。它的核心挑战在于消费速度和推理速度的匹配，后面我会详细讲。
第二种是批量离线分析模式。定时从 Kafka 里批量拉取一段时间内的消息，统一交给 AI 服务分析，结果落库或者生成报表。这种模式适合自己的分析系统，比如日报周报的自动生成、舆情趋势的周期性分析。它的好处是不要求 AI 服务的实时响应，可以挑业务低峰期跑，对资源的要求也低很多。
第三种是事件驱动的人机协同模式。AI 服务不是全自动处理每条消息，而是根据消息内容判断是否需要人工介入。比如内容审核场景，AI 先做预筛，只有置信度低的才推送给人工审核员。这种模式对准确率的要求比延迟更高，Kafka 在这里承担的是消息路由的职责，把不同置信度的消息分流到不同的下游。
我们的业务是三种模式混着用的：实时反馈走流式推理，历史分析走批量离线，审核兜底走人机协同。三种模式跑在同一个 Kafka 集群上，通过不同的 topic 做隔离，互不干扰。这种设计的好处是灵活，坏处是需要特别注意 topic 的规划，后面我会给出我们实际的 topic 划分方式。
2. 核心细节解析与实操要点
2.1 Kafka 环境准备：Docker 安装与可视化工具
如果是从零开始搭环境，我建议直接用 Docker 安装 Kafka，省去很多依赖折腾的时间。不过有一个关键点：Kafka 用到了 ZooKeeper 做集群协调，所以要先启动 ZooKeeper 再启动 Kafka。虽然新版 Kafka 引入 KRaft 模式可以摆脱 ZooKeeper，但考虑到大部分教程和线上集群还在用 ZooKeeper 方案，我这里还是以传统方式为例。
Docker Compose 是最快的启动方式。一个最小可用的配置大概长这样：
version: '3' services: zookeeper: image: bitnami/zookeeper:3.8 ports: - "2181:2181" environment: - ALLOW_ANONYMOUS_LOGIN=yes kafka: image: bitnami/kafka:3.0.0 ports: - "9092:9092" environment: - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181 - ALLOW_PLAINTEXT_LISTENER=yes - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092 - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 depends_on: - zookeeper注意KAFKA_CFG_ADVERTISED_LISTENERS这个参数，很多人第一次启动 Docker Kafka 后客户端连不上，基本都是因为这个。它告诉客户端“你应该用这个地址去连接 Kafka”，如果设置成了容器内部地址，宿主机上的客户端自然连不上。如果你要跨机器访问，这里要改成宿主机的实际 IP。
启动后建议装一个可视化管理工具，用来观察 topic、消息内容和消费者组的状态。我用的是 Kafka UI，一个开源项目，直接跑 Docker 镜像就行。它能帮你直观地看到每个 topic 的分区情况、每个消费者组的 lag（消费积压）情况，排查问题的时候非常有用。生产环境没有可视化工具的话，排查 lag 只能靠命令行盯着，效率低而且容易漏。
2.2 本地 AI 推理服务：Ollama 部署要点
本地大模型的部署，我们选 Ollama 有以下几个考量。第一，它把模型下载、量化、常驻内存这些细节全部封装好了，一条命令就能把模型跑起来；第二，它自动提供 OpenAI 兼容的 REST API，不需要额外写一层适配；第三，它对显存的利用比较聪明，默认会做量化加载，普通 24G 显存的卡就能跑 7B 到 13B 参数的模型，对设备要求不算离谱。
安装很简单，官网下载安装包或者用脚本一条命令装好。重点说下模型的选择。如果只是做文本分类、实体抽取、情感判断这类结构化任务，7B 模型完全够用；如果要生成连贯的长文本回复，13B 起步；如果是追求效果上限且显存吃紧，那就用量化版本。
模型跑起来之后，默认监听11434端口。验证是否就绪，可以执行：
curl http://localhost:11434/api/generate -d '{ "model": "qwen2.5:7b", "prompt": "这是一条测试消息", "stream": false }'如果返回了文本内容，说明服务正常。这里把stream设为 false，是为了让 API 一次性返回完整结果，而不是流式输出，流式输出对 Kafka 消费者这种程序化调用的场景不太友好。
2.3 客户端选型与消息格式约定
接入 Kafka 的客户端，我们同时用了 Python 和 Java，但消息格式是统一的 JSON。统一格式这件事非常关键，因为没有统一格式，消费者不知道如何解析消息内容，AI 服务的输入输出也就没法打通。
我们定义的统一消息格式大致是这样的：
{ "msg_id": "uuid", "source": "feedback_service", "content": "原始文本内容", "ext_info": { "user_id": "12345", "channel": "app" }, "timestamp": 1699344000000 }msg_id这个字段很重要，既用于去重，也用于后续结果回写时关联原始消息。content是交给 AI 服务的核心字段，ext_info放业务附带信息，AI 处理的时候可能需要参考，但本身不参与预测。
Python 客户端我们用的 confluent-kafka，消费的时候同时指定 topic 和 consumer group。这里要特别说明一下 group.id 的作用：同一个 group 里的消费者实例会分摊 topic 的分区，实现负载均衡。如果你起了多个消费者进程但 group.id 相同，它们会自动分配分区，不会重复消费同一条消息。后面讲重复消费问题时会再提到这个机制。
3. 实操过程与核心环节实现
3.1 整体链路流程设计
直接说我们最终跑通的链路。业务系统产生消息后，通过 Kafka 生产者把 JSON 消息发送到raw.user.feedbacktopic；AI 服务作为一个消费者程序，从该 topic 拉取消息，调用本地大模型推理，得到结果后把{msg_id, label, score, reply_text}封装成新消息发送到ai.result.feedbacktopic；下游的结果入库服务从ai.result.feedback消费，把结果落库。整个链路中 Kafka 是唯一的通讯通道，上游不关心 AI 服务是否存在，AI 服务也不关心下游怎么使用结果。
这个设计的核心价值在于，Kafka 让整条链路变成了完全异步的管道。业务系统的接口响应时间不受 AI 推理速度影响，AI 服务支持随时扩容缩容，下游消费速度慢时消息自动在 Kafka 里堆积而不会丢。用生活类比的话，Kafka 就像是生产者和消费者之间的一个超级邮箱，你往里投信就完事了，邮局（Kafka）保证信件不会丢，收信人什么时候来拿、怎么处理是收信人的事。
3.2 Python 消费者与 AI 推理实现
下面是消费者程序的核心代码骨架，大家可以参考这个结构做自己的适配：
import json import requests from confluent_kafka import Consumer, KafkaError AI_ENDPOINT = "http://localhost:11434/api/generate" def call_llm(content: str) -> dict: payload = { "model": "qwen2.5:7b", "prompt": f"请对以下用户反馈进行分类，判断情感极性并生成一条友好回复：\n{content}", "stream": False } resp = requests.post(AI_ENDPOINT, json=payload, timeout=60) resp.raise_for_status() data = resp.json() return { "label": data.get("label", "unknown"), "reply_text": data.get("response", "") } def main(): consumer_conf = { "bootstrap.servers": "localhost:9092", "group.id": "ai-feedback-group", "auto.offset.reset": "earliest", "enable.auto.commit": False } consumer = Consumer(consumer_conf) consumer.subscribe(["raw.user.feedback"]) while True: msg = consumer.poll(timeout=1.0) if msg is None: continue if msg.error(): if msg.error().code() == KafkaError._PARTITION_EOF: continue else: print(msg.error()) break data = json.loads(msg.value().decode("utf-8")) try: result = call_llm(data.get("content", "")) # 将结果发送到结果topic send_result(data["msg_id"], result) consumer.commit(asynchronous=False) except Exception as e: print(f"处理失败 msg_id={data['msg_id']}: {e}") # 这里根据业务需要决定是否commit，防止消息被反复消费 def send_result(msg_id: str, result: dict): # 生产端实现省略，关键是把内容发送到 ai.result.feedback pass if __name__ == "__main__": main()几个关键点需要特别说明。
一是enable.auto.commit: False加手动 commit。默认情况下 Kafka 消费者会自动提交 offset，但这会导致一个问题：消息被拉取后还没来得及处理完，offset 就提交了，程序崩溃时消息就丢了。手动 commit 是等消息处理成功后再提交，确保“至少一次交付”的语义。代价是可能重复消费，所以消息处理逻辑要支持幂等，也就是处理同一条消息多次不会产生脏数据。
二是auto.offset.reset: earliest。当消费者组第一次启动，或者 offset 失效时，这个参数决定了从哪个位置开始读。earliest 表示从最早的消息开始读，适合需要全量分析的场景；latest 表示只读新消息，适合实时流水处理场景。我建议在测试阶段用 earliest，确认稳定后再改成 latest 或者保持 earliest 并依赖手动提交机制。
三是超时设置。调用大模型接口一定要设置超时时间，否则模型服务卡住的时候，消费者进程会一直阻塞在那里，导致整个消费链路停滞。我们线上设置的是 60 秒，这个值需要根据模型的推理时间来调整。如果模型平均推理需要 10 秒，就把超时设成 30 秒，留足余量但又不能太长。
3.3 Spring Kafka YAML 配置：Java 侧接入
如果你是在 Spring Boot 项目里接入，YAML 配置比写代码更容易踩坑。我给出我们验证过的配置版本：
spring: kafka: bootstrap-servers: localhost:9092 producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.springframework.kafka.support.serializer.JsonSerializer properties: acks: all retries: 3 consumer: group-id: ai-feedback-group-java key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer auto-offset-reset: earliest enable-auto-commit: false properties: max-poll-records: 100 max-poll-interval-ms: 300000 listener: type: batchacks: all是最可靠的写入确认策略，生产者会等所有 ISR 副本都写入成功后才认为消息发送成功。代价是吞吐量略微下降，但对于 AI 相关业务，数据可靠性优先级明显更高。
max-poll-records表示消费者一次 poll 最多返回多少条消息。这个参数在 AI 场景下很有讲究，因为每条消息都要经过大模型推理，如果一次 poll 返回太多，处理时间就会很长，容易触发max.poll.interval.ms超时。我们线上设置为 100，配合批量监听器，一次处理一批，效率比单条处理高不少。
enable-auto-commit: false对应 Python 侧的手动提交逻辑。Spring Kafka 支持AckMode.MANUAL确认模式，处理完一批消息后手动调用 acknowledgment 方法提交 offset。
3.4 参数计算与性能正调优的思考过程
接入过程中最难的部分其实是性能调优，因为 AI 推理和 Kafka 消费速度天然不在一个量级。Kafka 单分区消费能力每秒几千条很轻松，但一个 7B 模型推理一条消息可能就要几百毫秒到几秒。这个速度差如果不处理好，Lag 会一直飙升。
我们从两个方向来解决。第一是提高并发度：AI 消费者进程可以部署多个副本，每个副本消费一个或多个分区，推理过程并行。第二是调整消费批次：不要让每条消息触发一次模型调用，而是攒一批消息后合并成一个大 Prompt 发给模型，让模型一次处理多条，减少 API 调用开销。
用具体的数字算一笔账。假设一个分区每秒进来 50 条消息，模型单条推理需要 1 秒。同步消费的话，一个消费者每秒只能处理 1 条，Lag 每秒增加 49 条，很快消息就积压成山了。但如果起 50 个消费者实例并行处理，每个消费者绑定一个分区，理论上每秒能处理 50 条，刚好追上生产速度。当然这是理想情况，实际受限于模型服务的并发能力和硬件资源。
批量处理的方式更激进一点。比如把 20 条消息拼成一个 Prompt，模型一次推理可能只需要 3 秒，那么平均每条消息耗时只有 0.15 秒，吞吐量提升了接近 7 倍。缺点是单次请求的响应时间变长，而且模型对长文本的处理效果可能有所下降。这个要靠实验找平衡点，没有统一答案。
我们最终采取的方案是：每条 Kafka 消息独立调用模型，但消费者进程起了 8 个副本，分布在两台机器上。这样单条消息的处理时延保持在 1 秒左右，吞吐量足够支撑当前的业务量。后续如果消息量翻倍，直接把副本数加一倍就行，Kafka 分区会自动负载均衡。
4. 常见问题与排查技巧实录
4.1 消息消费延迟高，Lag 持续上涨怎么办
这是 AI 接入 Kafka 最常见的故障。当你打开 Kafka UI 看到消费者的 lag 曲线一路上升，心里就该有数：消费速度跟不上生产速度了。先不要急着加机器，而是按下面的顺序排查。
第一步，检查是不是模型推理耗时异常。看消费者日志里单条消息的处理耗时，如果从原来的 1 秒涨到了 5 秒，那问题在模型侧。常见原因是显存不够导致模型开始走 CPU 计算，或者模型推理线程被其他任务抢占。
第二步，检查消费者有没有频繁 rebalance。rebalance 是指消费者组内成员发生变化时，Kafka 重新分配分区。每次 rebalance 期间分区是无人消费的，lag 会跳涨。看日志里如果有大量的rebalance关键字，就要检查消费者的max.poll.interval.ms是不是设得太短了，导致处理超时被误判为消费者退出。
第三步，检查分区数和消费者数的匹配。Kafka 的分区是消费并行的最小单位，一个消费者可以消费多个分区，但一个分区只能被同一个消费者组里的一个消费者消费。如果 topic 只有 3 个分区，你起了 10 个消费者，那也只能有 3 个消费者在干活，剩下的 7 个是空闲的。这种情况下要增加分区数才能提升并行能力。
4.2 重复消费与消息丢失问题
AI 业务对重复消费的容忍度需要分情况讨论。如果 AI 调用是纯幂等的（比如同样输入一定产生同样输出），重复消费无非是浪费一点算力。但如果 AI 服务有副作用，比如生成一条客服回复并推送出去，重复消费就意味着用户会收到两条一样的回复，这是不可接受的。
我们的做法是两条腿走路。一是消费端依靠 Kafka 的 offset 提交机制保证“至少一次”，二是业务端在消息里加 msg_id，在结果入库之前先查重，如果同样 msg_id 的结果已经存在就跳过。这样即使 Kafka 在某些极端情况下重复投递了消息，最终业务上也不会产生重复数据。
顺带提一下 Kafka 消费者组的一个特性：同一个 group 内的不同消费者实例消费的是不同分区的消息，但如果你新建了一个 group 从头消费同一个 topic，消息会被完整地“再消费一遍”。这是 Kafka 的基本语义，很多人会忽略。如果你想重新分析一遍历史数据，不改任何代码，只要换一个 group.id 重新运行消费者就行。这个特性在 AI 场景下非常有用，换更好的模型后重新跑历史数据就靠它。
4.3 大模型调用超时导致消费卡死的坑
这个问题我们踩得很深。最开始我们的消费者逻辑是：拉取消息，调用模型，等待返回，返回后再拉取下一条。某个周五下午模型服务因为显存溢出直接挂了，但消费者进程没有任何错误日志，只是安静地卡在等待响应的状态。等到我们发现的时候，Kafka 里的 lag 已经堆到了上百万条。
现在我们在每个环节都加了超时和熔断机制。调用模型的 HTTP 请求设了超时时间，超过直接抛异常；连续处理失败达到一定阈值后，消费者会进入“慢速模式”，暂停消费一段时间，让模型服务有喘息机会；同时加了告警，当 lag 超过设定的阈值时，自动通知值班人。这套机制之后，即使模型服务再出问题，最多也只是延迟几个小时，不会造成大面积堆积和雪崩。
4.4 Kafka 可视化工具箱：排查 Lag 的利器
排查问题的时候，可视化工具能帮你省一半的时间。推荐两个工具：Kafka UI 和 Offset Explorer。Kafka UI 是纯 Web 界面，功能全面，可以看到 topic 列表、分区详情、消费者组和 lag 情况，适合安装在一台共享服务器上供团队使用。Offset Explorer 以前叫 Kafka Tool，桌面客户端，适合个人快速连集群排查问题。
排查 lag 的时候，我一般先看 lag 绝对值，再看趋势。绝对值大不一定可怕，如果 lag 在逐渐下降说明消费端在追赶；稳中有升才说明消费能力确实不足。然后结合分区分布，看是不是某些分区的 lag 明显比其他分区高，这可能意味着某个 broker 节点有性能瓶颈，或者是某些消息特别难处理（比如特别长的文本），拖慢了整个分区的消费进度。
4.5 Kafka 与其他消息队列在 AI 场景的取舍
最后聊聊选型。我们内部也有人问，为什么不用 RabbitMQ 而用 Kafka？其实简单说，如果只是“消息通知 + 异步解耦”的简单场景，RabbitMQ 反而更合适，功能简洁，学习成本低。但我们的场景是“消息量大 + 需要回放 + 需要按 offset 精确控制消费进度”，这三个需求 RabbitMQ 都做得不太好。
Kafka 的核心优势在于日志化存储和分区的概念。分区让你可以水平扩展消费者能力，日志化存储让你可以随时从头或者从某个指定位置重新消费数据。AI 场景不仅要求消息能实时流转，还要求历史数据可回放、可重算，这正好打在 Kafka 的优势区上。RabbitMQ 消费确认后消息基本就消失了，想重新消费得靠额外的存储适配，非常麻烦。
还有一点是生态。Kafka 和 Flink、Spark、大模型框架的集成都比较成熟，无论是做实时特征工程还是离线数据回放，都有现成的连接器。做 AI 相关项目时，Kafka 已经不仅仅是消息队列了，更像是数据中枢，各种系统都从它这里拉数据和回写数据。
5. 经验总结与运维实践建议
5.1 接入 AI 后的 Kafka 运维新常态
Kafka 接入 AI 之后，运维的工作方式发生了明显变化。以前关注的是消息吞吐量、broker 负载、磁盘空间这类指标；现在还需要额外关注模型服务的状态、消费 lag 的走势、消息内容和模型输入格式的兼容性。任何一边出了问题，另一边的表现都很诡异。比如模型服务变慢，表现出来的是 Kafka lag 飙升；模型服务崩溃，表现出来的是消费者卡死和 lag 暴涨。如果你只盯着 Kafka 的指标找问题，很容易找错方向。
所以我们现在把 Kafka 和 AI 服务的监控指标放到了同一个看板上，用一条时间线同时展示两边的状态。发现问题时先看时间对应关系，比如 lag 开始涨的时间点和模型平均响应时间开始变长的点是否吻合，如果是，直接去查模型服务就好。如果不吻合，再去排查消费代码和网络链路。
5.2 监控指标与告警阈值的建议
下面是我们小组实践下来觉得比较合理的监控指标组合。
topic 维度的消息积压量是所有指标里必须最先看的。积压量级和业务容忍度直接相关，我们设定的规则是超过 10 万条持续 5 分钟就告警，因为按我们的消费能力，10 万条大约需要 30 分钟才能消化完，超过这个数意味着问题已经比较严重了。
消费端单条消息处理耗时也是核心指标。这个指标出现异常波动的意义比 lag 更早，lag 是结果，处理耗时是原因。我们每次发版前都会先跑一小批历史消息，看一下平均处理耗时和 P99 耗时，对性能走向有个预期。
AI 服务的成功率也要重视。大模型偶尔会产生幻觉、返回非 JSON 格式的内容，或者干脆报错，这些都会影响消费成功率。我们要求在代码里做严格的响应解析和异常捕获，解析失败的消息单独落入一个 dead letter topic，方便事后分析和重放。
5.3 一个小技巧：利用 Kafka 补偿机制做故障恢复
最后分享一个我们做得比较顺手的小技巧。AI 模型升级后，历史分析结果混沌，或者模型在某个时间段效果不佳，怎么快速重算？最简单的办法就是上面说过的“新 group 重放”：启动一个临时消费者，用新的 group.id 从最早 offset 消费目标 topic，把消息重新过一遍新模型，结果写入不同的结果 topic 或带上版本号。重放完成后，只要切换业务查询的版本号就行，不需要停线上服务和回滚数据。
这个能力在 Kafka 里是天然的，不需要任何额外组件。相比之下，RabbitMQ 或者其他基于消费确认删除的消息队列做这种历史重放就很麻烦。Kafka 的日志存储模型在这个场景下体现得淋漓尽致。我们靠着这个能力，把模型迭代的成本从“找数据、导数据、跑批、对账”这一整套流程，简化成了一条启动命令的事。
踩过几次坑之后，我个人对 Kafka 接入 AI 这个事最大的体会是：不要把精力都扑在写业务代码上，多花时间在设计消息格式、规划消费者组、调优参数和构建监控告警这些基础设施上。这些环节做得越扎实，后面业务跑起来就越顺手。特别是消息格式，一定要在一开始就把扩展字段预留好，不然等接入更多模型和场景后，改格式要牵动一大片代码，那种痛苦经历过才懂。
