对应ai-transfer项目的设计疑问

Question

这个智能调仓Agent有两个想请教一下老师
1.审核超时如果需要重发邮件机制或超时处理机制怎样设计？
2.如果如果调整建议，可能存在几个请求调用这个flow流程，而同一个调配类的物料，都卡在审批流程，审批员可能是多人处理（邮件组），怎样保证数据一致性，第一个审批通过，后面审批就失效防止数据错误。

Answer 1

Carl老师 2026-03-20 22:18:51

首先针对第一个问题，我们将agent进行了有状态设计，流程执行过程是保存在redis中的，如果你也可以自己去实现接口将执行流程保存在数据库，那么当你去进行审批的时候就能够成功从你持久化机制中召回流程。

不论是要重新触发发送邮件节点执行还是进行审批操作都需要借助于持久化，一单持久化状态没了，就无法实现了。

针对第二个问题，我们在kafka的消费时候进行了一个幂等处理，如果一个商品触发了库存调拨agent，只要没处理完这个商品的唯一key会放在redis中，只有处理完毕后redis中唯一key才会删除，对应的商品才可以再次触发，所以不会存在一个商品没有执行完毕的时候被重复触发

• 提问者 慕斯4245393 #1

  第二点，明白老师的意思（通过同key放在同一个分区或redis同key防止数据幂等性）。第一点不太明白你的意思，有类似的实现参考例子吗？因为审批调仓操作应该仓管团队负责，除了审批通过、审批不通过、还有一种超时的状态（没有跟进通知邮件），这会资源被占用因为是堵塞机制。我问了chatgpt，它推荐我在堵塞前使用缓存审批的信息，使用定时任务获取没有审批的信息，激活超时机制。

• Carl老师 回复 提问者 慕斯4245393 #2

  因为现在整个agent的执行过程都进行了持久化，哪怕你得邮件没有发出去，那么你可以在发送邮件之前将信息保存到数据库中的一张表，然后通过一个定时任务去比对agent执行过程中持久化的信息和数据库中发送邮件的数据，然后手动再次执行agent。
  而且spring ai alibaba graph不会因为一个结点阻塞就一直先用线程，它依然会适当线程。所以只要agent执行过程持久化不丢失，是可以重新在指定节点触发

• Carl老师 回复 提问者 慕斯4245393 #3

  首先我们课程中的整个graph是进行了redis持久化的，整个agent flow的执行情况都保存在了redis中。
  如果发送邮件之后，仓储人员一直不操作邮件，那么需要结合业务，比如一个调拨建议必须在2天内审批才可以，超时之后这个agent执行就结束。那么你发送邮件之后就等待仓储人员操作。为了防止他们不操作导致agent针对该商品的调拨不会重新给出建议的问题，那么你就需要写一个定时任务，每两天执行一次，执行逻辑则是去匹配redis中该threadId对应的agent没有执行完毕的工作流，然后将它从redis中删除。
  你也可以再写一个定时任务，逻辑是针对那些超过12小时没有操作邮件但是没有超过2天的流程，去执行邮件重新发送，（graph可以支持指定节点触发，你知道threadId还知道节点的id）
  另外不需要担心这个流程没有操作邮件就一直站着线程，你执行人类反馈的时候，这个节点graph会释放线程的，所以不会导致线程池中的线程被消耗殆尽的情况。

Answer 2

Carl老师 2026-03-20 22:20:59

kakfa的同一个队列是顺序消费的，哪怕多个请求同时触发同一个商品都往kafka发送了消息，但是消费的时候第一次消费就进行了幂等