DAG 与消息队列驱动的 AI 微服务

为什么需要 DAG

情报分析通常不是一次模型调用，而是多步处理：解析原文、识别显式和隐含标的、查询市场数据、过滤风险候选、执行最终优化。步骤之间存在依赖，但并非每条情报都经过同一条路径。DAG 将流程拆成节点并保存依赖关系，节点只通过上下文和消息通信，从而实现解耦、并行和故障隔离。

原始情报
  ├─▶ 显式提取 ─┐
  ├─▶ 账号找币 ──┼─▶ 结果合并 ─▶ 最终优化
  └─▶ 图片分析 ─┘

网关节点只负责路由，业务节点负责产生结果；这条边界使流程图可以变化，而单个 Handler 仍能独立测试。

消息生命周期

一条节点消息的生命周期可以概括为：

1. Consumer 从 Kafka 或 RabbitMQ 拉取消息。
2. 使用消息标识申请 Redis 幂等处理权。
3. 用 Pydantic 校验 NodeMessage，根据 slug 查找 Handler。
4. 创建与任务标识关联的 Langfuse Trace，执行异步 Handler。
5. 构造 NodeCallbackMessage，发送到回调队列。
6. 只有业务结果已发送且幂等状态原子变更为 done 后，才确认消息。

如果 Redis 不可用、回调发送失败或 processing 无法安全转换为 done，消费者不能静默确认消息，而应让消息重试。这样可以避免“业务副作用已经发生，但消息被当成成功消费”的不一致。

Kafka 与 RabbitMQ

该平台默认使用 Kafka 作为消息通道，同时保留 RabbitMQ 适配器。两者被统一到 Producer、Consumer 和消息类型抽象之后，上层 DAG 不需要感知底层实现。选择消息系统时，我关注的不是“哪个更流行”，而是：

• 是否支持当前的消费模型和并发度。
• 如何记录 offset、重试次数和积压情况。
• 故障时能否进入 DLQ（死信队列）。
• 监控指标是否能与业务节点关联。

Redis 幂等状态机

项目中的幂等状态不是简单地“查到就跳过”，而是一个带租约的状态机：

不存在 ──SET NX──▶ processing:<owner_token>
                         │
             完成成功 ───┴──▶ done
             执行失败 ───────▶ 删除 processing，允许重试
             租约过期 ───────▶ 其他消费者重新获取

processing 防止两个 Worker 同时执行，done 防止重复消息再次产生副作用。完成和释放通过 Redis Lua 脚本校验 owner token 后原子执行，避免旧 Worker 误删新 Worker 的租约。幂等 key 使用稳定的 namespace 和业务标识，不把物理队列名写进 key，从而降低队列改名造成的重复执行风险。

重试与 DLQ

可恢复的网络错误、服务暂时不可用和超时可以重试；参数校验、权限不足和资源不存在等确定性错误不应无限重试。超过重试次数后，消息进入 DLQ，并通过 worker_dlq_messages_total 等指标暴露出来。排障时需要同时查看：消息原文、重试计数、Worker 日志、节点 Trace 和 DLQ 原因。

我的收获

以前我把消息队列理解成“把任务放到后台执行”，在项目中才真正理解它还承担了流量削峰、服务解耦和失败恢复的责任。一个可靠的异步系统必须同时设计消息契约、幂等策略、确认时机、重试边界和监控指标，单独增加某一个组件并不能自动获得可靠性。

小结

DAG 解决“流程如何组合”，消息队列解决“节点如何异步协作”，Redis 幂等和 DLQ 解决“失败后如何恢复”。三者组合后，AI 处理链路才能在并发、重试和部分故障下保持可控。