第2章异步任务与队列¶

学习目标¶

理解 AI Agent 系统中异步任务的典型场景与价值
对比 RabbitMQ、Kafka、Celery 三类异步方案的区别与选型
掌握任务队列的核心设计：生产者、消费者、重试、幂等
理解消费者模式（竞争消费/广播/工作窃取）与死信队列
学会用任务监控保障异步链路的可观测性
能够用 Python + Celery 实现一个批量文档处理任务

问题导入¶

用户上传一份 200 页的 PDF，让 Agent 做"全文摘要 + 关键词抽取 + 章节问答生成"。这三步加起来要跑 5 分钟。如果用同步 HTTP 接口处理，会发生什么？

用户面前一直转圈，超时断开连接，任务白跑了。
服务器一个 worker 被这个长任务占住，无法服务其他用户。
中途服务重启，任务丢失，用户毫不知情。
100 个用户同时上传，100 个 worker 全被占满，系统卡死。

正确的做法是：HTTP 接口立刻返回一个 task_id，真正的处理丢到后台队列异步执行，用户通过轮询或 WebSocket 拿结果。这就是异步任务与消息队列要解决的问题。在 Agent 系统里，凡是"耗时超过 1 秒"的操作——批量 Embedding、长文档分块、多 Agent 协作、定时知识库刷新——都该走异步。

核心概念¶

1. 什么时候需要异步¶

判断标准有三：耗时（超过单次 HTTP 超时阈值）、可拆分（能分解为子任务并行）、可容忍延迟（用户接受"稍后查看结果"）。三者满足其一就该考虑异步。Agent 系统的典型异步场景：批量文档处理、定时 RAG 索引重建、多模型并行投票、长流程 Agent 工作流（如深度研究 Agent）。

2. 三大方案对比¶

维度	Celery	RabbitMQ	Kafka
定位	任务队列框架	通用消息中间件	分布式流平台
协议	AMQP/Redis 等	AMQP	自定义 TCP
持久化	依赖 broker	强持久化	日志追加，强持久
吞吐	中（万级 TPS）	中（万级 TPS）	极高（百万 TPS）
延迟	毫秒~秒	毫秒	毫秒
重试/死信	框架内置	需配置	需自建
适用	Python 任务编排	业务解耦、可靠投递	日志流、事件溯源、大数据

选型建议：纯 Python 任务编排选 Celery；多语言微服务解耦选 RabbitMQ；海量日志/事件流选 Kafka。三者并非互斥——Celery 的 broker 可以是 RabbitMQ 或 Redis，Kafka 也可作为事件源喂给 Celery 消费。

3. 任务队列核心设计¶

一个健壮的任务队列要解决五个问题：

投递保证：至少一次投递（At-Least-Once），通过消息持久化 + 消费确认实现。
幂等性：消费者可能收到重复消息，必须幂等。常用手段：业务唯一键去重、状态机校验、task_id + Redis 标记。
重试与退避：失败自动重试，但要有指数退避（避免重试风暴）和最大次数限制。
优先级与延迟：VIP 任务优先、定时任务延迟投递。
死信处理：重试耗尽的任务进死信队列，供人工介入。

4. 消费者模式¶

竞争消费（Competing Consumer）：多个消费者共享一个队列，每条消息只被一个消费。最常见，用于水平扩展。
发布订阅（Pub/Sub）：一条消息广播给所有订阅者。用于通知场景。
工作窃取（Work Stealing）：空闲消费者从忙碌消费者处"偷"任务，负载更均衡。Celery 的 prefetch 配合 acks_late 可近似实现。

5. 死信队列（DLQ）¶

消息被拒绝、TTL 过期、队列满时，进入死信队列。DLQ 是异步系统的"急救箱"：正常队列不被毒消息堵塞，坏消息集中可排查。Agent 系统中，LLM 返回异常内容、文档解析失败等"半坏"任务，都应进 DLQ 而非无限重试。

6. 任务监控¶

异步任务的"不可见"是最大风险。必须监控：队列积压长度、消费速率、任务 P95 延迟、失败率、DLQ 堆积。Celery 配合 Flower 可看实时任务状态；更通用的是把任务事件打到 Prometheus，用 Grafana 告警。

原理图¶

flowchart LR
    subgraph 生产者
        API[FastAPI 接口]
        SCHED[定时调度器]
    end

    API -->|提交任务| BROKER[(消息 Broker<br/>RabbitMQ/Redis)]
    SCHED -->|定时任务| BROKER

    BROKER --> Q1[普通队列]
    BROKER --> Q2[优先级队列]
    BROKER --> Q3[延迟队列]

    subgraph 消费者集群
        W1[Worker 1]
        W2[Worker 2]
        W3[Worker 3]
    end

    Q1 --> W1
    Q1 --> W2
    Q2 --> W3
    Q3 -.延迟到期.-> Q1

    W1 -->|执行| TASK[任务执行<br/>LLM/Embedding/解析]
    W2 --> TASK
    W3 --> TASK

    TASK -->|成功| RES[(结果后端<br/>Redis/DB)]
    TASK -->|失败+可重试| RETRY[重试<br/>指数退避]
    RETRY --> BROKER
    TASK -->|失败+耗尽| DLQ[(死信队列 DLQ)]

    DLQ -.->|告警| ALERT[运维告警]
    RES -.->|状态回调| API

    FLOWER[Flower/Prometheus] -.监控.-> W1
    FLOWER -.监控.-> W2
    FLOWER -.监控.-> Q1

图：异步任务队列架构。生产者把任务投递到 Broker，按类型路由到不同队列；消费者集群竞争消费，成功写结果后端，失败按策略重试或进死信队列；监控组件全程观测队列积压与任务状态。

示例代码¶

下面用 Celery 实现一个"批量文档处理"任务：分块 → Embedding → 入库，带重试、幂等、进度查询。

# tasks.py —— Celery 任务定义
import os
import time
import hashlib
from celery import Celery, group, chord
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

# 初始化 Celery，broker 用 Redis，结果后端也用 Redis
app = Celery(
    "agent_tasks",
    broker=os.getenv("CELERY_BROKER", "redis://localhost:6379/0"),
    backend=os.getenv("CELERY_BACKEND", "redis://localhost:6379/1"),
)

# 全局配置
app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="Asia/Shanghai",
    enable_utc=False,
    # 消费者：晚确认，任务执行完才 ack，避免 worker 崩溃丢任务
    task_acks_late=True,
    # 预取：每个 worker 一次只取 1 个，避免任务分配不均
    worker_prefetch_multiplier=1,
    # 重试：最大 3 次，指数退避
    task_default_max_retries=3,
    task_default_retry_backoff=True,        # 指数退避
    task_default_retry_backoff_max=300,      # 退避上限 5 分钟
    task_default_retry_jitter=True,          # 加随机抖动
)


@app.task(
    bind=True,
    name="agent.embed_chunk",
    autoretry_for=(Exception,),      # 任意异常自动重试
    retry_backoff=5,                  # 首次退避 5 秒
    retry_max=3,
)
def embed_chunk(self, doc_id: str, chunk_text: str, chunk_index: int) -> dict:
    """对单个文本块生成 Embedding 并入库。幂等：按 chunk 哈希去重。"""
    chunk_hash = hashlib.sha256(chunk_text.encode()).hexdigest()

    # 幂等检查：已处理过的 chunk 直接跳过
    if _already_embedded(chunk_hash):
        logger.info("chunk 已存在，跳过: %s", chunk_hash[:8])
        return {"doc_id": doc_id, "chunk_index": chunk_index, "cached": True}

    # 调用 Embedding API（可能失败，故可重试）
    vector = _call_embedding_api(chunk_text)
    _save_to_vector_db(doc_id, chunk_index, chunk_hash, vector, chunk_text)
    logger.info("chunk 已入库: doc=%s index=%d", doc_id, chunk_index)
    return {"doc_id": doc_id, "chunk_index": chunk_index, "cached": False}


@app.task(bind=True, name="agent.process_document")
def process_document(self, doc_id: str, chunks: list[str]) -> dict:
    """批量处理一篇文档：用 chord 编排「并行 embed → 汇总」。

    chord = group(子任务) | 回调任务
    - group: 所有 chunk 并行 embed
    - 回调:  全部完成后更新文档状态
    """
    # 启动 chord
    header = group(
        embed_chunk.s(doc_id, text, idx) for idx, text in enumerate(chunks)
    )
    callback = finalize_document.s(doc_id)
    chord(header)(callback)
    return {"doc_id": doc_id, "total_chunks": len(chunks), "status": "processing"}


@app.task(name="agent.finalize_document")
def finalize_document(results: list[dict], doc_id: str) -> dict:
    """所有 chunk 处理完后的回调：更新文档状态。"""
    cached = sum(1 for r in results if r.get("cached"))
    fresh = len(results) - cached
    _update_doc_status(doc_id, "completed")
    logger.info("文档处理完成: doc=%s 新增=%d 缓存=%d", doc_id, fresh, cached)
    return {"doc_id": doc_id, "status": "completed", "fresh": fresh, "cached": cached}


# ---- 模拟依赖 ----
def _already_embedded(chunk_hash: str) -> bool:
    """查 Redis 判断是否已入库。"""
    import redis
    r = redis.Redis.from_url("redis://localhost:6379/2", decode_responses=True)
    return r.exists(f"emb:{chunk_hash}") == 1

def _call_embedding_api(text: str) -> list[float]:
    """调用 Embedding API，返回向量。"""
    from openai import OpenAI
    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    resp = client.embeddings.create(model="text-embedding-3-small", input=text)
    return resp.data[0].embedding

def _save_to_vector_db(doc_id, idx, chunk_hash, vector, text):
    import redis
    r = redis.Redis.from_url("redis://localhost:6379/2", decode_responses=True)
    r.hset(f"doc:{doc_id}:chunks", idx, text)
    r.set(f"emb:{chunk_hash}", "1", ex=86400)  # 标记已处理

def _update_doc_status(doc_id, status):
    import redis
    r = redis.Redis.from_url("redis://localhost:6379/2", decode_responses=True)
    r.hset(f"doc:{doc_id}", "status", status)

下面是 FastAPI 接口，提交任务并查询进度：

# api.py —— FastAPI 接口
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from celery.result import AsyncResult
from tasks import app as celery_app, process_document

api = FastAPI()

class DocRequest(BaseModel):
    doc_id: str
    chunks: list[str]

@api.post("/documents/process")
def submit_document(req: DocRequest):
    """提交文档处理任务，立即返回 task_id。"""
    task = process_document.delay(req.doc_id, req.chunks)
    return {"task_id": task.id, "status": "queued", "total_chunks": len(req.chunks)}

@api.get("/tasks/{task_id}")
def get_task_status(task_id: str):
    """查询任务状态与结果。"""
    result = AsyncResult(task_id, app=celery_app)
    response = {"task_id": task_id, "state": result.state}
    if result.ready():
        response["result"] = result.result
    return response

启动方式：

# 1. 启动 worker（消费任务）
celery -A tasks worker --loglevel=info --concurrency=4

# 2. 启动 FastAPI
uvicorn api:api --reload

# 3. 启动 Flower 监控（可选）
celery -A tasks flower --port=5555

企业案例¶

案例 1：批量文档处理¶

某法律科技公司每天需处理上万份合同：OCR → 分块 → Embedding → 入库 → 摘要生成。单份合同处理约 40 秒，串行需 11 小时。团队用 Celery 改造：每份合同的分块用 group 并行 embed，多份合同间用 4 个 worker（concurrency=16）并行消费。改造后日处理量提升到 10 万份，且单份失败可单独重试而非整批回滚。acks_late=True 保证 worker 崩溃不丢任务，DLQ 收集解析失败的合同供人工排查。

案例 2：多 Agent 协作的异步编排¶

一家研究机构用多 Agent 做行业研报：搜索 Agent、分析 Agent、写作 Agent 串行协作，全流程 20 分钟。用同步 HTTP 调用容易因网络抖动中断。改为 Celery 任务链（chain）：每步结果存结果后端，下一步从后端取输入，任何一步失败自动重试。配合 Flower 监控，研究员可实时看到"搜索完成→分析进行中"。

常见问题¶

Q1：消息会丢吗？ 会，除非你做对三件事：broker 开持久化（Redis 用 RDB/AOF，RabbitMQ 队列设 durable）、消息设持久化（delivery_mode=2）、消费者用 acks_late（执行完才确认）。三者缺一不可。

Q2：怎么保证幂等？ 核心是"业务唯一键 + 去重"。示例中用 chunk_hash 做幂等键，已处理的 chunk 标记在 Redis。也可用数据库唯一约束、状态机（只处理特定状态的任务）。

Q3：Celery 的 prefetch 该设多少？ 长任务（秒级以上）设 worker_prefetch_multiplier=1，避免一个 worker 预取太多导致其他 worker 空闲。短任务（毫秒级）可调大（如 4-8）减少网络往返。

Q4：任务积压怎么办？ 短期：加 worker。长期：拆分队列（紧急/普通分离）、限流生产端、优化任务本身（如批量合并 Embedding 请求）。监控积压长度并告警，别等用户投诉才发现。

Q5：RabbitMQ vs Redis 做 Celery broker？ Redis 简单快但持久化弱（崩盘可能丢几秒数据），适合容忍少量丢失的场景；RabbitMQ 持久化强、支持优先级队列和 DLX（死信交换机），适合对可靠性要求高的生产环境。

面试题¶

1. 为什么消费者要做幂等？ 答：消息中间件保证 At-Least-Once 而非 Exactly-Once，网络抖动或重平衡时消费者会收到重复消息。若不幂等，重复消费会导致重复入库、重复扣款等问题。幂等手段：唯一键去重、状态机、乐观锁。

2. acks_late=True 解决什么问题？ 答：默认消费者一收到消息就 ack，若执行中崩溃，消息已丢失。acks_late 改为执行完成后才 ack，崩溃时消息会重新入队，交给其他消费者。代价是必须做好幂等。

3. 死信队列的意义？ 答：重试耗尽的消息进 DLQ，避免毒消息无限重试堵塞正常队列，同时集中坏消息便于人工排查。DLQ 是异步系统可靠性的最后一道防线。

4. Celery 的 group、chain、chord 区别？ 答：group 并行执行多个任务（扇出）；chain 串行执行，前一个输出作后一个输入（流水线）；chord = group + 回调，所有并行任务完成后触发一个汇总任务（扇出→汇总）。

5. Kafka 与 RabbitMQ 在 Agent 系统中如何取舍？ 答：Kafka 适合高吞吐日志流、事件溯源（如所有 Agent 对话日志汇聚分析）；RabbitMQ 适合任务分发与业务解耦（如"处理这份文档"的可靠投递）。Agent 系统常二者并用：Kafka 收日志，RabbitMQ 跑任务。

练习题¶

在示例基础上，给 embed_chunk 增加优先级：VIP 用户的文档优先处理（提示：Celery 支持多队列，或 RabbitMQ 优先级队列）。
实现一个"延迟任务"：提交后 10 分钟才开始执行（提示：Celery countdown 或 eta，或 Redis 有序集合自建延迟队列）。
配置 Flower 并观察：故意让一个任务失败，看重试与退避的曲线；把 retry_max 改为 1，观察任务如何进入失败状态。
思考题：如果 Embedding API 突然限流（每秒只允许 5 个请求），如何在消费者侧做"全局限流"避免被 ban？

小结¶

本章聚焦 AI Agent 系统的异步任务与消息队列。核心要点：凡耗时、可拆分、可容忍延迟的任务都该异步化；Celery 适合 Python 任务编排，RabbitMQ 适合可靠投递，Kafka 适合事件流；健壮的任务队列必须解决投递保证、幂等、重试退避、死信、监控五件事。我们用 Celery 实现了批量文档处理的 chord 编排，展示了并行 embed + 汇总的完整模式。下一章进入系统的"眼睛"——日志、监控与链路追踪，让异步链路里发生的一切变得可观测。

第2章 异步任务与队列¶