Agent 任务队列架构优先级、死信队列和状态追踪一、个性化深度引言Agent系统上线第三周运营同学在群里贴了张截图为什么用户说帮我整理上周的报销Agent回复了正在查询天气我翻开日志才发现两个不同用户的请求在消息队列中被绑错了上下文。更严重的是一个需要调用支付接口的高风险任务因为网络抖动重试了3次导致重复扣款。Agent不是简单的请求-响应模型。一个Agent任务可能涉及多步推理、多次工具调用、异步等待外部API、以及复杂的错误恢复。任务队列的设计决定了Agent的可靠性和一致性。二、个性化原理剖析Agent任务队列比普通消息队列多三个特殊需求优先级调度安全操作高于信息查询、状态追踪用户需要知道任务进行到哪一步、以及死信处理失败任务不能被静默丢弃。stateDiagram-v2 [*] -- PENDING: 任务创建 PENDING -- QUEUED: 入队成功 QUEUED -- EXECUTING: 调度执行 EXECUTING -- WAITING_TOOL: 调用外部工具 WAITING_TOOL -- EXECUTING: 工具返回 EXECUTING -- SUCCESS: 执行完成 EXECUTING -- RETRYING: 可恢复错误 RETRYING -- EXECUTING: 重试 RETRYING -- DEAD_LETTER: 超过最大重试 EXECUTING -- FAILED: 不可恢复错误 WAITING_TOOL -- DEAD_LETTER: 工具超时 SUCCESS -- [*] FAILED -- [*] DEAD_LETTER -- MANUAL_REVIEW: 人工介入 DEAD_LETTER -- ARCHIVED: 确认丢弃见证奇迹的时刻在于状态机的设计。我们将Agent任务的状态从常见的3~4个扩展为8个其中 WAITING_TOOL 是最关键的一个中间状态。Agent调用外部API时任务进入等待状态并释放计算资源。当API返回时任务从断点继续而非从头执行。这种设计将一个10步推理链条的GPU占用从持续15秒降低为间断占用累计3秒集群吞吐量提升了近5倍。优先级设计分为三级P0涉及支付、权限变更等不可逆操作——永远优先处理且不允许自动重试超3次P1查询类操作——常规优先级允许重试P2分析类后台任务——可在GPU闲时处理。三、个性化代码实践from enum import Enum import asyncio import json from dataclasses import dataclass, field from datetime import datetime from typing import Any, Optional class TaskState(Enum): PENDING pending QUEUED queued EXECUTING executing WAITING_TOOL waiting_tool # 新增状态等待工具返回 RETRYING retrying SUCCESS success FAILED failed DEAD_LETTER dead_letter class TaskPriority(Enum): P0_CRITICAL 0 # 支付、删除等不可逆操作 P1_NORMAL 1 # 常规查询 P2_BACKGROUND 2 # 分析类后台任务 dataclass class AgentTask: task_id: str user_id: str instruction: str priority: TaskPriority TaskPriority.P1_NORMAL state: TaskState TaskState.PENDING retry_count: int 0 max_retries: int 3 tool_call_history: list field(default_factorylist) created_at: datetime field(default_factorydatetime.now) # 设计原因checkpoint 记录执行断点支持工具返回后从断点继续 checkpoint: Optional[dict] None class AgentTaskQueue: Agent 任务优先级队列 def __init__(self): # 设计原因三个独立优先队列避免高优任务被低优任务阻塞 self.p0_queue asyncio.PriorityQueue() self.p1_queue asyncio.PriorityQueue() self.p2_queue asyncio.PriorityQueue() # 设计原因死信队列持久化到磁盘防止进程重启丢失 self.dead_letter_queue DeadLetterStore() # 设计原因在内存中维护状态表供状态查询 API 使用 self.task_registry: dict[str, AgentTask] {} async def enqueue(self, task: AgentTask): 入队 task.state TaskState.QUEUED self.task_registry[task.task_id] task # 设计原因用创建时间戳作为优先级值同优先级FIFO priority_value task.created_at.timestamp() queue_map { TaskPriority.P0_CRITICAL: self.p0_queue, TaskPriority.P1_NORMAL: self.p1_queue, TaskPriority.P2_BACKGROUND: self.p2_queue, } await queue_map[task.priority].put((priority_value, task)) async def dequeue(self) - Optional[AgentTask]: 按优先级出队 # 设计原因总是先检查P0队列保证关键任务优先 for queue in [self.p0_queue, self.p1_queue, self.p2_queue]: if not queue.empty(): _, task await queue.get() return task return None async def handle_failure(self, task: AgentTask, error: Exception): 失败处理区分可重试和不可重试错误 if self._is_retryable(error) and task.retry_count task.max_retries: task.state TaskState.RETRYING task.retry_count 1 # 设计原因指数退避避免对故障服务造成更大压力 delay min(2 ** task.retry_count, 30) # 最大30秒 await asyncio.sleep(delay) await self.enqueue(task) else: task.state TaskState.DEAD_LETTER await self.dead_letter_queue.store(task) # 设计原因P0任务进入死信立即告警其他级别仅记录 if task.priority TaskPriority.P0_CRITICAL: await self._alert_oncall(task) def _is_retryable(self, error: Exception) - bool: 判断错误是否可重试 # 设计原因网络超时可重试业务逻辑错误不可重试 # 幂等性由上层保证请求带了 idempotency_key retryable_types (TimeoutError, ConnectionError) return isinstance(error, retryable_types)四、个性化边界权衡优先级反转问题P2任务占用GPU时新来的P0任务需要等待当前任务完成。如果P2任务涉及大模型多步推理耗时20秒P0任务的等待时间不可接受。解决方案是引入抢占机制——当P0任务到达时如果当前执行的是P2任务且已运行超过3秒则保存checkpoint后暂停先执行P0任务。状态一致性的挑战Agent的中间状态如WAITING_TOOL分布在队列、Worker内存和外部工具之间。Worker宕机时WAITING_TOOL状态的任务可能丢失上下文。需要将状态持久化到Redis且给WAITING_TOOL状态设置心跳超时——超时未收到工具返回时任务自动回退到QUEUED状态重新调度。死信队列的容量线上每天产生约0.3%的死信任务。如果不定期清理半年后死信队列数据量可能达到TB级。需要设置过期策略P1/P2任务死信保留7天P0任务死信保留30天。幂等性不是队列的责任队列保证at-least-once投递但Agent的任务幂等性需要业务层保证。尤其是P0任务支付、删除必须在指令级别携带idempotency_key。五、总结Agent任务队列需要支持优先级调度、8状态状态机和死信处理三个核心能力。优先级防止关键任务被阻塞详细状态追踪让用户感知任务进展死信队列确保失败任务不被丢失。引入WAITING_TOOL中间状态可将计算资源利用率提升数倍。抢占机制和状态持久化是工程落地必须考虑的额外设计。
