当前位置：首页 > news >正文

Agent 任务队列架构：优先级、死信队列和状态追踪

news 2026/7/18 23:35:59

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_factory=list) created_at: datetime = field(default_factory=datetime.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。