Agent 服务的并发与状态：异步、任务队列、幂等和故障恢复怎么配合

一条 Agent 请求可能等待模型、检索、工具和审批。请求多起来以后，问题不仅是“能开多少协程”，还有任务会不会丢、写入会不会重复、旧 worker 会不会覆盖新结果，以及客户端断开后系统怎样继续。

本文综合官方运行框架、FastAPI、PostgreSQL 与 outbox 资料。代码片段是用于解释语义的简化示例；完整生产实现还要接入鉴权、存储、调度、取消、监控和故障演练。

两家厂商保存的状态，与应用状态如何连接

OpenAI SDK Sessions^[1]解决跨运行对话历史；人工审批机制^[2]支持暂停、保存运行状态并继续。这些能力帮助恢复模型交互，但某个工具是否已改变业务对象，仍需要业务系统确认。

Claude Managed Agents^[3]公开了 session、harness 和 sandbox 的解耦思路。对应用的启发是把可靠事件记录与可替换的执行进程分开；该工程报告并没有给所有第三方工具定义事务或幂等保证。

工程推导：即使用厂商托管状态，应用也应保存业务任务 ID 到厂商 session / run ID 的映射，以及已确认结果与验收状态。重复回调、超时后的查询和用户取消，都必须落到同一个业务任务上。

async I/O：提高等待期间的利用率

FastAPI 并发指南^[4]解释了并发等待与并行计算的区别。等待模型 HTTP、异步数据库或网络工具时，await 可以释放事件循环；同步库放进 async def 并不会自动变异步，反而可能阻塞所有同进程请求。

同步 I/O 可以使用适当的线程池与超时；较重的 Python CPU 计算通常应使用独立进程或专门计算服务，具体还受运行时与原生库行为影响。PDF 解析、OCR、embedding 批处理与交互式流式响应可使用不同 worker 池，避免批任务占满交互资源。

下面展示进程内限制模型调用的语义，invoke_model 是待接入的异步函数：

model_slots = asyncio.Semaphore(16)  # 示例值，按压测和上游预算调整

async def bounded_call(request, remaining_seconds):
    # 等待并发槽也消耗截止时间，不能只给 HTTP 调用计时。
    async with asyncio.timeout(remaining_seconds):
        async with model_slots:
            return await invoke_model(request)

这段代码不提供多进程或跨机器的总限流。启动四个进程后可能允许四倍调用；需要全局配额时，由入口或共享调度器分配预算，并防止所有实例同时重试。asyncio.timeout 也无法中断阻塞事件循环的同步计算。

容量：协程数量不是吞吐承诺

在稳态平均条件下，Little 定律给出 在途任务数 ≈ 到达率 × 平均停留时间。示例推算：接受 20 个任务/秒、平均停留 15 秒，则平均约有 300 个任务处于系统中；这不是 300 个模型调用都应同时执行，也不是 p95 容量保证。

一个任务可能调用模型三轮、并发检索两次，或暂停等审批。容量规划分别估计：HTTP 连接、运行中任务、模型调用 RPM/TPM、数据库连接、工具并发和队列积压。限制应覆盖用户、租户、模型和工具，防止一个慢租户耗尽公共资源。

队列等待过长时应拒绝新任务或返回明确的繁忙状态；可以按优先级与公平性调度。扩 worker 前先看瓶颈：若上游 token 配额已满，扩容只会增加排队与限流。

持久队列：接收成功必须有恢复依据

FastAPI BackgroundTasks^[5]适合响应后的进程内工作，重型工作可以使用专门队列。它本身不能替代需要持久化、重试与跨进程恢复的任务系统。

接收长任务时，推荐在一个数据库事务中写入任务与待发布事件，再返回 202 + task_id。发布器把 outbox 事件送到消息队列，确认后标记发布；worker 依据任务 ID 去重消费。AWS transactional outbox^[6]解释了如何避免数据库提交与消息发送之间的双写漏洞，也指出重复消息仍需消费者处理。

图 1 · 队列允许重投；任务接收、领取和业务动作分别有稳定身份。外部结果未知时先对账。

接口幂等键按租户或用户范围唯一，并绑定规范化请求摘要。同一键同一内容返回原任务；同一键不同内容明确拒绝。先查“是否存在”再插入仍有竞争，应通过唯一约束或原子写入完成。

最小数据模型：哪些数据由谁负责

MySQL / PostgreSQL 可保存事务事实；Redis 可做缓存、短期配额或调度辅助；消息队列负责传输；向量库负责检索索引。没有必要强行使用每一种组件，也不要把可能淘汰的缓存当作任务成功的唯一记录。

若业务确实有设备、监护关系或其他角色，在身份与授权模型上扩展关系；它们属于具体业务约束，不应硬编码进通用 Agent 循环。

领取任务与 fencing：租约过期不代表旧进程已停止

worker 用短事务领取任务，记录租约到期时间和递增的执行代次 epoch，事务提交后再调用模型或工具。不能持有数据库行锁等待几分钟的外部请求。

PostgreSQL SELECT 文档^[7]说明 FOR UPDATE SKIP LOCKED 可用于多个消费者访问队列式表，但返回的是跳过锁定行的视图，不适合普通一致查询。它也不提供长期任务租约或重试策略，仍需要应用设计。

任务更新可以使用条件写入：WHERE task_id = ? AND epoch = ? AND owner = ?，并结合有效租约校验，只有当前执行者能提交状态。旧进程恢复后提交旧代次结果，应被拒绝。

局限在外部副作用：数据库 fencing 只能挡住数据库所验证的写入。工具服务若不验证代次、稳定操作 ID 或幂等条件，旧 worker 仍可能执行外部动作。关键写入通过受控工具网关校验，或依赖上游唯一约束与查询对账，不能只加一个 Redis 锁就宣布“恰好执行一次”。

三个崩溃窗口：如何避免盲目重跑

尝试 ID 每次执行可以变化；逻辑操作 ID 跨重试保持稳定。同一个操作若参数或目标版本改变，应被识别为新动作并重新授权，不能复用旧幂等键掩盖差异。未知结果没有查询能力时，暂停并升级处理比自动重试更可靠。

流式接口、慢客户端与取消

HTTP 接收确认、Agent 运行和事件流可以分开：客户端订阅任务事件，用事件序号或 SSE ID 恢复展示；服务器定义重放窗口，超出窗口时先读取任务快照再接续。确认、进度与模型文本是不同事件，不能把普通进度称为首 token。

慢客户端会使缓冲增长。对可重放事件设置有限缓冲和明确断开策略；实时模型流若无法暂停或重放，定义中止与续接行为。不要因为展示端慢就无限占用内存，也不能静默丢掉用户需要的最终状态。

取消是一项持久化请求：worker 在模型、工具和步骤边界检查，将取消尽量传播上游。已提交外部动作不能靠断开 SSE 撤销；还要处理“取消与完成同时发生”的状态竞争，依据已确认事实返回结果。多端协议与旧客户端兼容见端云同步篇。

优化与验证：用故障和负载测出边界

固定模型、任务组合和上游配额，逐步调整 worker 数量、并发上限与批大小，记录接受率、队列等待、p95、工具超时、数据库连接和每成功任务费用。负载包含长短任务混合、单租户突发、慢消费者与429，不能只测平均延迟。

故障演练主动终止 worker、重复投递消息、丢失外部响应、让租约过期后旧 worker 恢复，并在运行中滚动升级。验收关注任务不丢、确认不重、越权不发生、未知结果能对账；这些是业务保证，不能由“用了 async + Redis + MQ”自动推出。

参考资料