AI 应用从 Demo 走向生产环境最容易被低估的一环不是模型能力而是数据流转的实时性。过去两年大家拼的是能不能跑通现在拼的是跑得稳不稳、快不快、准不准。我所在的团队最近半年一直在做 AI 应用的生产化改造踩过的坑几乎都集中在实时数据智能这一块——不是模型不行而是数据到得不够快、不够对、不够全。这篇文章就把我们趟出来的经验完整拆开讲从架构选型到异步通信从 Agent 编排到云原生部署尽量把每个决策背后的为什么说清楚。1. 为什么实时数据智能成了 AI 生产化的分水岭1.1 从能回答到答得对的鸿沟早期做 AI 应用大家关注的是模型能不能理解问题、能不能生成像样的回答。这个阶段用离线数据、批量灌入知识库就能应付用户问一个问题系统去向量库里检索几段文本拼进 Prompt 里让模型生成效果看起来还不错。但一旦进入真实生产场景问题就暴露了用户问的是现在库存还有多少这个订单当前状态是什么刚才那笔交易有没有异常这些问题的答案每秒钟都在变离线数据根本喂不上。我们内部做过一个统计在客服场景里超过六成的用户问题涉及实时状态查询。如果 AI 回答的是十分钟前的数据用户第一次可能觉得还行第二次就会直接找人工。这不是模型能力问题是数据链路问题。实时数据智能要解决的核心矛盾就是模型推理是秒级的但数据供给如果还是分钟级甚至小时级整个系统的价值就会大打折扣。1.2 实时性带来的连锁反应实时数据一旦接入整个系统的复杂度会指数级上升。离线场景下你可以容忍数据延迟、可以批量重跑、可以事后修正。但实时场景下数据是流式的、连续的、有时序的任何一个环节的抖动都会传导到最终回答上。我们最初的做法是让 AI 应用直接查业务数据库简单粗暴但很快就遇到了三个问题一是高频查询把业务库压得喘不过气二是数据库的 schema 是面向事务设计的不是面向检索的查询效率很低三是多个 Agent 并发查询时连接池瞬间打满。这三个问题逼着我们去重新设计数据层。后来我们的思路是业务库不动在它和 AI 应用之间加一层实时数据管道用变更数据捕获CDC把业务库的变更实时同步到一个专门面向检索的存储里AI 应用只查这一层。这个改动看起来简单但它是整个实时数据智能架构的基石。1.3 什么样的场景真正需要实时数据智能不是所有 AI 应用都需要实时数据。我见过不少团队一上来就追求全实时结果架构复杂度飙升收益却不明显。判断标准其实很简单如果数据的时效性直接影响用户的决策或体验那就需要实时如果数据晚几分钟甚至几小时对结果没影响那就没必要上实时链路。具体来说以下几类场景对实时性的要求最高交易风控类数据延迟直接意味着资金风险智能客服类用户问的是当前状态答错会直接导致投诉运维监控类异常检测需要秒级响应供应链调度类库存和物流状态变化频繁调度决策依赖最新数据。反过来像知识问答、内容生成、代码辅助这类场景对实时性的要求就低得多用离线知识库加定期更新完全够用。2. 实时数据管道的搭建从 CDC 到检索层的完整链路2.1 变更数据捕获的选型与取舍实时数据管道的第一步是把业务库的变更捕获出来。市面上主流的方案有三种基于数据库日志的 CDC、基于触发器的 CDC、基于应用双写的 CDC。我们最终选了基于日志的方案具体来说是解析数据库的 binlog。原因很直接对业务库侵入最小不需要改表结构不需要加触发器性能损耗可以控制在百分之几以内。基于触发器的方案我们早期试过问题是每次写操作都要额外触发一次写入高并发下延迟明显而且触发器逻辑一旦出问题很难排查。应用双写的方案更不可取它要求业务代码同时写两个地方一致性完全靠应用层保证一旦有一边写失败就会出现数据不一致而且对业务代码的侵入太大。基于日志的方案也不是没有坑。最大的坑是 schema 变更。业务库加个字段、改个类型CDC 管道如果没同步处理就会解析失败或者丢数据。我们的做法是在 CDC 层加一个 schema 注册中心所有 schema 变更必须先注册再上线管道根据注册信息动态适配。这个机制上线后因为 schema 变更导致的数据问题基本归零。2.2 消息队列在管道中的角色CDC 捕获到的变更不能直接写进检索层中间需要一个缓冲和分发层这就是消息队列的作用。我们用的是 Kafka核心考虑是三点高吞吐、可持久化、支持多消费者。高吞吐不用多说业务高峰期每秒几万条变更很常见可持久化是为了防止下游故障时数据丢失Kafka 可以把消息保留几天甚至几周下游恢复了再消费多消费者是为了让同一份变更数据能同时供给多个下游比如一个消费者写检索层一个消费者做实时特征计算一个消费者做审计归档。这里有个经验值得分享Kafka 的 topic 分区数不是越多越好。我们一开始为了追求并行度把分区数设得很大结果发现消费者端的 rebalance 变得非常频繁每次 rebalance 都会导致短暂的消费停顿。后来我们把分区数控制在消费者数量的两到三倍rebalance 频率明显下降整体吞吐反而更稳定。分区数的设置要结合消费者数量和单条消息的处理耗时来算不能拍脑袋。2.3 检索层的设计为什么不用向量库直接扛很多人一提到 AI 应用的检索层第一反应就是向量数据库。但实时数据智能场景下纯向量库是不够的。原因在于实时数据查询往往是结构化条件加语义检索的混合查询。比如找出过去一小时内在华东地区发生的、金额超过一万的、且描述类似退款纠纷的订单这里面既有时间范围、地区、金额这些结构化条件又有语义相似度匹配。我们的做法是分层存储结构化条件走倒排索引或列式存储语义检索走向量索引查询时先做结构化过滤缩小候选集再在候选集上做向量检索。这样既保证了召回率又控制了延迟。如果直接用向量库扛全部查询结构化过滤只能在向量检索之后做候选集太大会导致延迟飙升。实测下来分层方案在千万级数据量下P99 延迟能控制在两百毫秒以内而纯向量方案在同样数据量下经常超过一秒。3. Agent 编排中的异步通信别让同步调用拖垮整个系统3.1 同步调用的隐性成本Agent 架构刚流行的时候大家的做法很朴素一个主 Agent 接到任务依次调用工具 Agent、检索 Agent、生成 Agent每一步都是同步等待。这种模式在 Demo 阶段没问题但生产环境下问题很大。假设一个任务需要调用五个子 Agent每个子 Agent 平均耗时五百毫秒同步模式下总耗时就是两秒半。如果其中某个子 Agent 因为下游依赖抖动变成两秒整个任务就变成四秒。用户等四秒才看到第一个字体验直接崩掉。更严重的是资源占用。同步调用意味着主 Agent 的线程在整个等待期间都被占着不能处理其他请求。并发一上来线程池瞬间打满新请求只能排队。我们压测时发现同步模式下单实例并发超过五十就开始出现明显排队而异步模式下同样实例能扛到三百以上。3.2 异步通信的几种落地方式异步通信不是简单地把同步调用改成异步就完事了它涉及整个调用链的重构。我们实践下来主要有三种落地方式各有适用场景。第一种是消息队列解耦。主 Agent 把子任务作为消息投递到队列子 Agent 消费消息、处理、再把结果投递到结果队列主 Agent 从结果队列里收结果。这种方式解耦最彻底子 Agent 可以独立扩缩容某个子 Agent 挂了也不影响其他。缺点是链路变长端到端延迟会增加而且需要处理消息的顺序和幂等。第二种是响应式编程。用 Reactor 或者类似框架把调用链组织成数据流主 Agent 订阅子 Agent 的结果流有结果就处理没结果就等着不阻塞线程。这种方式延迟低适合对响应时间敏感的场景。缺点是对开发者的心智负担比较重调试起来不如同步代码直观。第三种是事件驱动加状态机。主 Agent 维护一个任务状态机每个子 Agent 完成后发一个事件状态机根据事件推进任务状态。这种方式最适合长流程、多步骤的任务比如需要人工审批介入的流程。缺点是状态管理复杂需要考虑状态持久化和恢复。我们最终是混合使用的短链路、低延迟要求的用响应式长链路、需要解耦的用消息队列涉及人工介入的用状态机。没有银弹关键是看场景。3.3 超时、重试与降级的实战配置异步通信绕不开超时、重试和降级这三个问题。我们的配置原则是超时时间要分层设置重试要有上限和退避降级要有兜底方案。超时分层的意思是不同层级的调用设置不同的超时。比如主 Agent 调用子 Agent 的超时是两秒子 Agent 调用下游服务的超时是八百毫秒下游服务调用数据库的超时是两百毫秒。这样任何一层出问题都能在上一层超时之前暴露出来避免雪崩。我们最初所有层都设五秒超时结果一个慢查询能把整条链路拖死。重试的策略是只对幂等操作重试重试次数不超过三次每次重试间隔指数退避。非幂等操作比如写操作重试可能导致重复写入我们改成先查后写或者用唯一键约束来保证幂等。重试间隔从一百毫秒开始每次翻倍最多到一秒。这样既能应对瞬时抖动又不会在持续故障时疯狂重试把下游压垮。降级的兜底方案分几档如果实时数据查不到降级到查最近一次的快照数据并在回答里标注数据可能有延迟如果子 Agent 完全不可用降级到只返回主 Agent 能处理的部分结果如果整个链路都挂了返回一个友好的错误提示并引导用户稍后重试。降级的关键是让用户感知到系统还在工作而不是直接报错。4. 云原生部署下的资源博弈GPU 配额、沙盒与弹性伸缩4.1 GPU 配额管理的现实困境AI 应用上云原生GPU 配额是最现实的约束。我们遇到过好几次GPU 配额已不够预冻结的报错任务提交上去直接被拒。这个问题的根源在于GPU 是稀缺资源云平台的配额是硬上限而 AI 应用的 GPU 需求波动很大——推理高峰期需要大量 GPU低谷期又闲置。我们的应对策略有三条。第一是推理和训练分离训练任务用抢占式实例能接受被中断推理任务用预留实例保证稳定性。第二是模型量化把 FP16 量化到 INT8显存占用直接减半同样的 GPU 能跑更多实例。第三是动态批处理把多个推理请求攒成一批一起送进 GPU提高 GPU 利用率。这三条组合下来我们的 GPU 成本降了将近四成。4.2 沙盒环境的安全边界Agent 执行代码或者调用外部工具时沙盒是必须的。我们用的是容器级沙盒每个 Agent 任务跑在独立的容器里有独立的文件系统、网络命名空间和资源限制。这样即使 Agent 执行了恶意代码也影响不到宿主机和其他任务。沙盒配置里有几个参数特别关键。CPU 和内存限制不用多说超了直接 OOM 或者被 throttle。网络策略要严格默认禁止所有出站连接只白名单必要的服务。文件系统要挂载成只读需要写入的目录单独挂载临时卷任务结束就销毁。还有一个容易被忽略的是执行时间限制我们设的是单任务最长五分钟超时直接 kill。这个限制防止了死循环或者卡死的任务长期占用资源。4.3 弹性伸缩的触发条件设计云原生的弹性伸缩听起来很美但触发条件设计不好反而会导致频繁扩缩容系统稳定性下降。我们的经验是扩容要快缩容要慢。扩容的触发条件用两个指标CPU 利用率超过百分之七十或者请求队列长度超过阈值。两个条件满足任意一个就扩容扩容步长是当前实例数的百分之五十最多不超过配额上限。扩容要快是因为流量高峰来得猛慢一步用户就排队了。缩容的触发条件用三个指标同时满足CPU 利用率低于百分之三十、请求队列为空、且持续五分钟以上。三个条件都满足才缩容缩容步长是当前实例数的百分之二十。缩容要慢是因为流量低谷可能只是暂时的缩太快了下一个高峰又得扩来回抖动反而浪费资源。5. Agent 记忆与多 AI 协作的工程化落地5.1 短期记忆与长期记忆的分层Agent 记忆是最近讨论很多的话题但很多实现只停留在把对话历史塞进上下文这个层面。生产环境下这种做法很快会遇到上下文长度限制和成本问题。我们的做法是分层短期记忆存最近几轮对话直接进上下文长期记忆存关键事实和用户偏好用向量库存储按需检索。短期记忆的窗口大小要权衡。窗口太小Agent 记不住上下文回答会前后矛盾窗口太大token 消耗高而且模型对长上下文的注意力会稀释。我们实测下来最近十轮对话是个比较平衡的点超过十轮的信息就压缩成摘要存进长期记忆。长期记忆的写入要有选择性不是什么信息都值得存。我们的规则是用户明确表达的偏好、任务的关键结论、需要跨会话保持的状态这三类才写入长期记忆。其他信息用完就丢。这样既控制了存储成本又保证了检索时的信噪比。5.2 多 Agent 协作的通信协议多 Agent 协作不是把几个 Agent 凑在一起就行它们之间需要一套通信协议。我们用的是基于消息的协议每个 Agent 有唯一的标识消息包含发送者、接收者、消息类型、负载和关联 ID。关联 ID 用来把同一个任务的多条消息串起来方便追踪和调试。消息类型我们定义了几种任务派发、结果返回、状态查询、错误上报、心跳。任务派发和结果返回是主要的状态查询用于主 Agent 监控子 Agent 的健康状况错误上报用于子 Agent 主动通知异常心跳用于检测子 Agent 是否存活。这套协议看起来简单但它让整个多 Agent 系统变得可观测、可调试出问题时能快速定位是哪个 Agent 的哪一步出了岔子。5.3 协作中的冲突处理多 Agent 协作最麻烦的是冲突。比如两个 Agent 同时对同一份数据做了修改或者两个 Agent 给出了矛盾的建议。我们的处理原则是能预防的预防预防不了的就仲裁。预防的手段主要是加锁和版本控制。对共享资源的写操作先获取分布式锁写完释放。对共享数据的修改带上版本号版本不匹配就拒绝写入让 Agent 重新读取最新版本再操作。仲裁的手段是设一个主 Agent 作为协调者当子 Agent 之间出现矛盾时由主 Agent 根据预设规则裁决比如以最新数据为准或者以置信度高的结果为准。6. 生产环境下的可观测性与故障排查6.1 实时数据链路的监控指标实时数据智能系统的监控不能只看 CPU 和内存这些基础指标更要看数据链路的健康度。我们重点监控几个指标数据延迟也就是变更发生到检索层可查的时间差这个指标直接反映实时性数据积压也就是消息队列里待消费的消息数积压说明下游处理不过来数据一致性定期抽样比对业务库和检索层的数据确保没有丢失或错乱。数据延迟我们设的告警阈值是五秒超过就告警。实测下来正常情况下延迟在五百毫秒以内网络抖动时会到一两秒超过五秒基本就是出问题了。数据积压的告警阈值根据业务量动态调整一般是正常消费速率的十分钟量。数据一致性的检查是每小时跑一次抽样一万条比对不一致率超过万分之一就告警。6.2 Agent 执行链路的问题定位Agent 执行链路长出问题时定位困难。我们的做法是全链路追踪每个任务生成一个 trace ID从主 Agent 到子 Agent 到下游服务每一跳都带上这个 ID日志和指标都按 trace ID 聚合。这样排查时拿到一个 trace ID 就能看到整个任务的完整执行路径哪一步慢、哪一步错一目了然。常见的 Agent 执行问题有几类超时通常是下游依赖慢或者网络抖动死循环通常是 Agent 的决策逻辑有 bug反复调用同一个工具上下文溢出通常是短期记忆窗口设得太大或者对话轮次太多工具调用失败通常是参数格式不对或者下游服务不可用。每一类问题我们都有对应的排查手册新人照着手册走基本能定位到根因。6.3 从故障中恢复的实战流程故障恢复的关键是快和稳。快是指快速止损稳是指恢复后不再复发。我们的流程是发现故障后先切降级方案保住核心功能然后定位根因修复后灰度恢复观察一段时间再全量。举个例子有一次检索层因为数据量突增导致查询超时AI 应用大面积报错。我们的处理是第一步把检索层的查询超时从两百毫秒放宽到一秒先让请求能返回第二步临时扩容检索层实例分担压力第三步定位到是某个大客户的批量导入导致数据量突增和客户协调错峰导入第四步给检索层加了写入限流防止类似情况再发生。整个过程从发现到恢复用了不到十五分钟核心功能没有中断。7. 一些踩坑之后的个人体会做实时数据智能这一年多最大的体会是实时不是目的可靠才是。很多团队为了追求实时性把架构搞得极其复杂结果稳定性反而下降。我的建议是先想清楚业务到底需要多实时是秒级、分钟级还是小时级然后按需设计不要过度工程。第二个体会是异步通信虽然好但不是所有地方都适合。短链路、低延迟要求的场景同步调用反而更简单可靠。异步带来的复杂度只有在链路长、并发高、需要解耦的场景下才划算。第三个体会是可观测性要提前做不要等出问题了才补。我们最初没做全链路追踪排查一个问题要翻好几台机器的日志后来补上追踪之后排查效率提升了不止一个量级。监控和追踪这些基础设施越早投入回报越高。最后一个体会是关于 GPU 和云资源的。云原生虽然弹性好但配额是硬约束不要假设资源随时可得。关键任务要有预留资源非关键任务才用抢占式。而且资源使用要有配额管理防止某个任务把配额吃光导致其他任务无法提交。这些都是真金白银换来的教训。