【免费下载链接】FinSight-AIAI equity research agent with resilient workflows, evidence-grounded RAG, versioned reports, and automated quality evaluation.项目地址https://gitcode.com/gh_mirrors/fi/FinSight-AI点击查看免费下载FinSight AI 是一个 AI 股票研究 Agent它把财报采集 → 指标计算 → 文档索引 → 情报构建 → AI 分析组织成一条长时运行工作流。这篇文章带你完整理解它的超时恢复WorkflowRecoveryScheduler与死信队列Dead Letter Queue机制当某个 Agent 任务卡死、崩溃或反复失败时系统如何自动识别、重试最终把无法恢复的任务优雅地停放到死信区。为什么 AI Agent 需要超时恢复与死信队列AI 工作流天然昂贵且易失败检索可能超时、模型调用可能挂起、索引可能只完成一半。如果任务卡死在RUNNING状态且无人处理就会出现沉默的部分失败——用户永远等不到结果资源也被白白占用。FinSight 的设计目标很明确让每个昂贵的阶段都是幂等的、可恢复的、可观测的。完整设计思路可参考 docs/design-agent-workflow.md。任务状态机先看懂 7 种状态在讲恢复之前先认识 WorkflowStatus.java 定义的状态机状态含义CREATED任务已创建等待执行RUNNING已获取租约正在执行RETRYING失败后等待重新入队SUCCEEDED执行成功FAILED本次失败可重试DEAD_LETTER重试次数耗尽进入死信区COMPENSATING补偿中预留状态每个任务还额外记录attempts尝试次数、errorMessage、leaseOwner和fencingToken防重令牌见 WorkflowTask.java。核心机制一WorkflowRecoveryScheduler 定时扫描超时任务超时恢复的入口是 WorkflowRecoveryScheduler.java它只做三件事扫描每recovery-delay-ms默认 60 秒扫描一次数据库找出所有状态为RUNNING且updatedAt早于当前时间 − running-timeout默认 10 分钟的任务。标记调用task.recoveredAfterTimeout(...)根据attempts次数决定去向——尝试次数 ≥ 3 直接标记DEAD_LETTER否则标记FAILED并写入错误信息Workflow timed out at stage XXX。重发布若标记后状态为FAILED可重试则将任务转为RETRYING通过WorkflowTaskPublisher重新投递到消息队列。关键细节是第 38 行的saveIfOwned它用状态 fencingToken 的比较并置位compare-and-set来更新任务。这意味着即使旧工人其实还活着比如只是 GC 停顿导致心跳变慢它的过期写入也会被 fencingToken 挡住不会覆盖恢复结果。配套索引保证扫描高效CREATE INDEX idx_workflow_tasks_status_updated ON workflow_tasks(status, updated_at ASC);见 V8__workflow_recovery_and_report_versions.sql。核心机制二RabbitMQ 侧的重试队列与死信交换器调度器只负责发现与标记真正把任务重新跑起来靠的是 RabbitMQ 的队列拓扑定义在 RabbitWorkflowConfiguration.java。这里有三组队列配合工作主队列finsight.workflow.ingestion绑定x-dead-letter-exchange消息被拒绝reject后进入死信交换器。重试队列finsight.workflow.retry设置x-message-ttl retry-delay-ms默认 5 秒。消息在这里休眠5 秒后过期再由 TTL 死信机制自动路由回主队列——这是一个用TTL DLX 实现的延时重试无需额外的定时器线程。死信队列finsight.workflow.dead-letter最终无法消费的消息停在这里方便人工排查与重放。消费端的失败处理见 RabbitWorkflowListener.java任务执行抛异常后若数据库状态为FAILED则转为RETRYING并发布到重试队列publishRetry否则重新抛出异常让消息走死信交换器。配合setDefaultRequeueRejected(false)被拒绝的消息不会原地反复重投避免热循环。这样形成了完整的三层失败处理执行异常 → FAILED尝试3次→ RETRYING → 重试队列延时5s→ 主队列 执行异常 → 尝试≥3次 → DEAD_LETTER任务层死信停止重试 消费拒绝 → 消息层死信交换器 → finsight.workflow.dead-letter队列层死信核心机制三Redis 租约 Fencing Token 防止重复执行恢复机制有个隐含前提任务必须幂等且同一任务不能被两个执行者同时处理。FinSight 用 Redis Lua 脚本实现单飞租约见 RedisBackedWorkflowLeaseService.java获取租约原子地检查键是否存在 → 递增 fencing 计数器 → 写入owner:token带 TTL续租/释放Lua 脚本校验owner:token一致后才操作防止误删他人租约。WorkflowOrchestrator.java 还会以TTL/3 的间隔发送租约心跳finsight.workflow.lease.renewal.total指标记录 renewed / ownership_lost / failed 三种结果确保长任务在 10 分钟恢复窗口内始终活着。本地开发没有 Redis 时租约服务会回退到进程内ConcurrentHashMap单飞表allow-local-lease-fallback: true生产环境关闭回退、fail-closed保证跨实例互斥不被静默绕过。关键配置参数速查表所有参数集中在 application.yml 的finsight.workflow段参数默认值作用running-timeoutPT10MRUNNING 状态超过该时长未更新即判定超时recovery-delay-ms60000恢复扫描间隔毫秒retry-delay-ms5000重试队列消息 TTL即重试延迟dead-letter-queuefinsight.workflow.dead-letter队列层死信队列名allow-local-lease-fallbacktrue是否允许本地租约回退生产建议false另外每次恢复都会打点finsight.workflow.recovery.total带taskType与stage标签让卡死任务从隐性问题变成可监控、可告警的显性指标。总结FinSight AI 的可靠性设计可以浓缩为一句话状态持久化 租约互斥 定时恢复 队列死信。WorkflowRecoveryScheduler.java 负责发现卡死、标记失败、触发重投RabbitWorkflowConfiguration.java 用 TTL 队列 DLX 实现延时重试与两层死信任务层DEAD_LETTER 消息层死信队列RedisBackedWorkflowLeaseService.java 用 Lua 租约与 fencingToken 保证恢复过程不会与仍在执行的旧工人冲突。这套组合拳让 AI Agent 的长时工作流在模型超时、进程崩溃、消息积压等常见故障下都能自动恢复或安全停下而不是把失败藏在沉默里。赞分享【免费下载链接】FinSight-AIAI equity research agent with resilient workflows, evidence-grounded RAG, versioned reports, and automated quality evaluation.项目地址https://gitcode.com/gh_mirrors/fi/FinSight-AI点击查看免费下载相关推荐如何构建高效的AI Agent异步消息队列Agentic完整指南如何构建高效的AI Agent异步消息队列Agentic完整指南 在当今AI驱动的应用开发中构建高效的异步消息处理系统是提升AI Agent性能的关键。AgAI AgentAPI网关后端开发工具终极指南ThingsBoard消息死信队列DLQ实现失败消息处理与恢复的完整方案终极指南ThingsBoard消息死信队列DLQ实现失败消息处理与恢复的完整方案 在物联网IoT平台中消息的可靠传递至关重要。ThingsBoard物联网后端数据可视化微服务消息队列Logstash 队列与数据韧性Queues and Data Resiliency完全指南内存队列、持久队列与死信队列的选型、配置与故障恢复Logstash 队列与数据韧性Queues and Data Resiliency完全指南内存队列、持久队列与死信队列的选型、配置与故障恢复 Logst数据工程后端创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考