第 17 章 · SQS 消息队列与 DynamoDB

本章目标：
理解 SQS Standard 与 FIFO 两种队列的区别及适用场景；
掌握 boto3 对 SQS 的核心操作：创建队列、发送/接收/删除消息；
理解死信队列（DLQ）的作用与配置方式；
学会用 DynamoDB 持久化 Agent 的状态检查点；
能将 SQS 消费者与 DynamoDB 组合成一个完整的生产级 Agent Worker。

本章是生产部署的关键一环。前面的章节（FastAPI ch22 讲了 Docker 基础）已经覆盖了容器化，这里我们讨论"容器之外"的基础设施——让 Agent 服务能够异步、可靠、水平扩展地运行。

17.1 SQS 标准队列与 FIFO 队列

Amazon Simple Queue Service（SQS）是最简单的 AWS 托管消息队列。Agent 系统用 SQS 的典型场景是：HTTP 接口接收用户请求 → 把任务塞进队列 → 后台 Worker 异步处理 → 结果写回数据库。

SQS 提供两种队列类型：

特性	Standard 队列	FIFO 队列
顺序保证	尽力而为，偶尔乱序	严格有序，同一 MessageGroupId 内顺序不变
吞吐量	极高（>10,000 条/秒）	较低（≤3,000 条/秒）
重复保证	至少一次	恰好一次（去重窗口 5 分钟）
适用场景	大量无关任务（如日志处理、通知）	有顺序要求的任务（如工单状态流转）
队列名后缀	无前缀	必须以 `.fifo` 结尾

💡 选型原则：Agent 任务是"独立且可重试"的 → Standard；Agent 需要严格按顺序处理（如同一用户的多个操作不能乱序）→ FIFO。

用 boto3 创建队列

python

import boto3

sqs = boto3.client('sqs', region_name='us-east-1')

# Standard 队列：默认属性
standard_queue = sqs.create_queue(
    QueueName='agent-tasks-standard',
    Attributes={
        # 消息保留 7 天（最大），防止消息丢失
        'MessageRetentionPeriod': '604800',
        # 消费者处理一条消息最长 300 秒，超时后消息重新入队
        'VisibilityTimeout': '300',
    }
)
print("Standard 队列 URL:", standard_queue['QueueUrl'])

python

# FIFO 队列：必须指定 DeduplicationScope 和 FifoThroughputLimit
fifo_queue = sqs.create_queue(
    QueueName='agent-tasks.fifo',
    Attributes={
        'FifoQueue': 'true',
        'ContentBasedDeduplication': 'true',  # 用消息体做去重（推荐）
        'MessageRetentionPeriod': '604800',
        'VisibilityTimeout': '300',
    }
)
print("FIFO 队列 URL:", fifo_queue['QueueUrl'])

ContentBasedDeduplication：设为 true 时，SQS 自动用消息体的 SHA-256 作为去重键，开发者无需自己传 MessageDeduplicationId。优先使用此方式，比手动管理更简单且不易出错。

17.2 发送与接收消息

发送消息

Standard 队列发送（无需 GroupId）：

python

from datetime import datetime

response = sqs.send_message(
    QueueUrl=standard_queue['QueueUrl'],
    MessageBody='{"task":"rag_query","question":"什么是 LangGraph？"}',
    # 可选：延迟交付（秒）
    DelaySeconds=0,
    # 可选：消息标签（用于后续搜索）
    MessageAttributes={
        'task_type': {'DataType': 'String', 'StringValue': 'rag_query'},
        'priority':   {'DataType': 'Number',   'StringValue': '1'},
    }
)
print("消息 ID:", response['MessageId'])

FIFO 队列发送（必须提供 GroupId）：

python

response = sqs.send_message(
    QueueUrl=fifo_queue['QueueUrl'],
    MessageBody='{"task":"update_ticket","ticket_id":"TK-1001","status":"closed"}',
    MessageGroupId='ticket-TK-1001',   # 同一工单的消息严格按顺序
    # 若 ContentBasedDeduplication=true，可不传 DeduplicationId
)
print("消息 ID:", response['MessageId'])

接收消息（轮询）

python

response = sqs.receive_message(
    QueueUrl=standard_queue['QueueUrl'],
    MaxNumberOfMessages=10,           # 一次最多取 10 条
    VisibilityTimeout=30,             # 处理期间其他消费者看不到（覆盖队列级别默认值）
    WaitTimeSeconds=20,               # 长轮询：最多等 20 秒再返回
    AttributeNames=['All'],           # 返回所有属性（含 ApproximateReceiveCount）
    MessageAttributeNames=['All'],    # 返回所有消息属性
)

messages = response.get('Messages', [])
for msg in messages:
    print(f"收到消息: {msg['Body']}")
    print(f"  已被重试次数: {msg['Attributes']['ApproximateReceiveCount']}")
    print(f"  标签: {msg['MessageAttributes']}")

长轮询（Long Polling）：WaitTimeSeconds=20 是关键——如果队列此时为空，SQS 会挂起连接最多 20 秒，等有消息到达时才响应。这避免了空轮询产生的请求费用，生产环境务必设置。

处理完毕后删除消息

python

for msg in messages:
    try:
        # 执行业务逻辑（可能抛异常）
        process_agent_task(json.loads(msg['Body']))
    except Exception as e:
        # 处理失败 —— 不删除消息，等待 VisibilityTimeout 后重新入队
        print(f"处理失败，将重新入队: {e}")
        continue
    # 处理成功，显式删除
    sqs.delete_message(
        QueueUrl=standard_queue['QueueUrl'],
        ReceiptHandle=msg['ReceiptHandle']
    )

ReceiptHandle 是 SQS 每次返回消息时分配的一次性令牌，删除时必须带上它，防止误删其他消费者的消息。

17.3 死信队列（Dead Letter Queue）

消息处理反复失败时，如果不加以控制，它会无限次重新入队，占用队列并掩盖真正的问题。死信队列（DLQ） 是一个专门的"墓地队列"——当源队列中的消息重试超过阈值后，SQS 自动把它移到 DLQ，源队列不再尝试投递。

python

# 第一步：先创建 DLQ
dlq_queue = sqs.create_queue(
    QueueName='agent-tasks-dlq',
    Attributes={'MessageRetentionPeriod': '1209600'}  # 保留 14 天
)
dlq_arn = dlq_queue['QueueArn']  # 记住 ARN，用于配置源队列

# 第二步：在源队列上配置 RedrivePolicy
sqs.set_queue_attributes(
    QueueUrl=standard_queue['QueueUrl'],
    Attributes={
        'RedrivePolicy': f'''{{
            "deadLetterTargetArn": "{dlq_arn}",
            "maxReceiveCount": 3
        }}'''
    }
)

配置生效后，一条消息被 receive_message 拉出 3 次且都未删除，第 4 次可见时 SQS 会自动移入 DLQ，源队列中的可见性条目消失。

python

# 定期巡检 DLQ：找出卡住的任务人工介入
dlq_messages = sqs.receive_message(
    QueueUrl=dlq_queue['QueueUrl'],
    MaxNumberOfMessages=10,
    WaitTimeSeconds=5,
)
for msg in dlq_messages.get('Messages', []):
    body = json.loads(msg['Body'])
    print(f"DLQ 卡住的 tasks: type={body.get('task')} | "
          f"原始队列重试了 {msg['Attributes']['ApproximateReceiveCount']} 次")
    # 可以选择：修复后 re-queue 或标记失败
    sqs.delete_message(
        QueueUrl=dlq_queue['QueueUrl'],
        ReceiptHandle=msg['ReceiptHandle']
    )

💡 最佳实践：任何生产队列都必须配 DLQ。否则"卡住的消息"会淹没你的监控和日志。

17.4 DynamoDB：Agent 状态存储

Agent 系统在运行时需要持久化状态：某个会话（thread）走到哪一步了、上次的推理结果是什么、有没有正在进行的工具调用……这些信息存在 SQLite 里不适合生产，存内存里一重启就丢。DynamoDB 是 AWS 托管的 NoSQL 键值库，毫秒级延迟、无限扩展，天然适合当 Agent 的"大脑记忆"。

建表

python

import boto3
from boto3.dynamodb.types import TypeDeserializer

dynamodb = boto3.resource('dynamodb', region_name='us-east-1')

table = dynamodb.create_table(
    TableName='agent-checkpoints',
    KeySchema=[
        {'AttributeName': 'pk', 'KeyType': 'HASH'},   # 分区键
        {'AttributeName': 'sk', 'KeyType': 'RANGE'},  # 排序键
    ],
    AttributeDefinitions=[
        {'AttributeName': 'pk', 'AttributeType': 'S'},
        {'AttributeName': 'sk', 'AttributeType': 'S'},
    ],
    BillingMode='PAY_PER_REQUEST',  # 按请求付费，无需预估容量
)
table.meta.client.get_waiter('table_exists').wait(TableName='agent-checkpoints')
print("表已就绪")

表的键设计遵循"一个 thread 对应一条记录"的原则：

属性	值示例	说明
`pk`	`thread#th-abc123`	会话 ID，同 thread 的状态集中存放
`sk`	`checkpoint#0`	检查点序号，支持历史回溯
`type`	`checkpoint` / `result`	区分状态快照与最终结果
`data`	`{"messages":[...]}`	JSON 序列化的完整状态
`updated_at`	`2026-06-30T10:00:00Z`	时间戳，用于乐观锁

写入检查点（PutItem）

python

def save_checkpoint(thread_id: str, step: int, state: dict) -> None:
    """把 Agent 当前状态持久化到 DynamoDB"""
    item = {
        'pk':         f'thread#{thread_id}',
        'sk':         f'checkpoint#{step}',
        'type':       'checkpoint',
        'data':       state,                         # DynamoDB 原生支持嵌套 dict
        'updated_at': datetime.utcnow().isoformat(),
    }
    table.put_item(Item=item)
    print(f"✅ 已保存 thread={thread_id} step={step}")

读取最新检查点（Query）

python

def get_latest_checkpoint(thread_id: str):
    """查询某个 thread 的最新检查点"""
    response = table.query(
        KeyConditionExpression='pk = :pk',
        ExpressionAttributeValues={':pk': f'thread#{thread_id}'},
        ScanIndexForward=False,   # 降序，第一条就是最新的
        Limit=1,
    )
    items = response.get('Items', [])
    return items[0]['data'] if items else None

ScanIndexForward=False + Limit=1 是一种常用的"取最新一条"模式——DynamoDB 不支持 ORDER BY ... LIMIT 的标准 SQL 写法，这个组合能达到同样效果。

条件更新（乐观锁）

多 Worker 同时更新同一个 thread 时，用 ConditionExpression 做乐观锁：

python

def update_checkpoint(
    thread_id: str, step: int, new_state: dict, expected_version: str
) -> bool:
    """只有在 updated_at == expected_version 时才写入，防止覆盖"""
    try:
        table.put_item(
            Item={
                'pk':         f'thread#{thread_id}',
                'sk':         f'checkpoint#{step}',
                'type':       'checkpoint',
                'data':       new_state,
                'updated_at': datetime.utcnow().isoformat(),
            },
            ConditionExpression='updated_at = :expected',
            ExpressionAttributeValues={':expected': expected_version},
        )
        return True
    except dynamodb.meta.client.exceptions.ConditionalCheckFailedException:
        print(f"⚠️ thread={thread_id} 版本冲突，跳过本次更新")
        return False

ConditionalCheckFailedException 不是错误——是正常的并发保护机制，业务代码应按预期处理（重试或放弃）。

17.5 完整示例：SQS 驱动 + DynamoDB 持久化的 Agent Worker

把上面两部分合在一起，就是一个生产级 Agent Worker 的骨架：

python

"""
agent_worker.py
SQS 消息驱动 + DynamoDB 状态持久化的 Agent Worker
"""
import json
import logging
import boto3
from boto3.dynamodb.conditions import Key
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

SQS_QUEUE_URL = "https://sqs.us-east-1.amazonaws.com/123456789/agent-tasks-standard"
DLQ_QUEUE_URL = "https://sqs.us-east-1.amazonaws.com/123456789/agent-tasks-dlq"

sqs = boto3.client('sqs', region_name='us-east-1')
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table('agent-checkpoints')


def process_rag_query(thread_id: str, question: str) -> dict:
    """模拟 RAG Agent 处理流程"""
    logger.info(f"[{thread_id}] 开始处理问题: {question[:50]}...")

    # Step 1: 从 DynamoDB 恢复之前状态（断点续传）
    existing = get_latest_checkpoint(thread_id)
    state = existing or {"messages": [], "retrieved_docs": []}

    # Step 2: 执行 Agent 逻辑（这里用模拟代替真正的 LLM 调用）
    state["messages"].append({"role": "user", "content": question})
    state["messages"].append({"role": "assistant", "content": f"（模拟）关于「{question[:20]}」的答案…"})
    state["retrieved_docs"].append({"doc_id": "doc-001", "score": 0.92})

    # Step 3: 持久化到 DynamoDB
    step = len(state["messages"]) // 2
    save_checkpoint(thread_id, step, state)

    logger.info(f"[{thread_id}] 处理完成，共 {len(state['messages'])} 条消息")
    return state


def get_latest_checkpoint(thread_id: str) -> dict | None:
    resp = table.query(
        KeyConditionExpression=Key('pk').eq(f'thread#{thread_id}'),
        ScanIndexForward=False,
        Limit=1,
    )
    items = resp.get('Items', [])
    return items[0]['data'] if items else None


def save_checkpoint(thread_id: str, step: int, state: dict) -> None:
    table.put_item(Item={
        'pk':         f'thread#{thread_id}',
        'sk':         f'checkpoint#{step}',
        'type':       'checkpoint',
        'data':       state,
        'updated_at': datetime.utcnow().isoformat(),
    })


def main():
    logger.info("🚀 Agent Worker 启动")
    while True:
        response = sqs.receive_message(
            QueueUrl=SQS_QUEUE_URL,
            MaxNumberOfMessages=5,
            VisibilityTimeout=60,
            WaitTimeSeconds=10,
        )
        messages = response.get('Messages', [])
        if not messages:
            continue

        for msg in messages:
            body = json.loads(msg['Body'])
            thread_id = body.get('thread_id', 'unknown')
            task_type = body.get('task')

            try:
                if task_type == 'rag_query':
                    result = process_rag_query(thread_id, body['question'])
                    # 把结果写回 DynamoDB 作为 final_result
                    table.put_item(Item={
                        'pk': f'thread#{thread_id}',
                        'sk': 'result',
                        'type': 'result',
                        'data': result,
                        'completed_at': datetime.utcnow().isoformat(),
                    })
                logger.info(f"✅ 处理完毕 thread={thread_id}")
            except Exception as e:
                logger.error(f"❌ 处理失败 thread={thread_id}: {e}")
                # 不 delete_message，让 VisibilityTimeout 到期后重新入队
                continue

            # 成功才删除
            sqs.delete_message(QueueUrl=SQS_QUEUE_URL, ReceiptHandle=msg['ReceiptHandle'])

        # 检查 DLQ，输出告警
        dlq_resp = sqs.receive_message(QueueUrl=DLQ_QUEUE_URL, MaxNumberOfMessages=10, WaitTimeSeconds=1)
        dlq_msgs = dlq_resp.get('Messages', [])
        if dlq_msgs:
            logger.warning(f"⚠️  DLQ 有 {len(dlq_msgs)} 条消息，请立即排查")
            for dm in dlq_msgs:
                logger.warning(f"   DLQ msg: {dm['Body'][:100]}")
                sqs.delete_message(QueueUrl=DLQ_QUEUE_URL, ReceiptHandle=dm['ReceiptHandle'])


if __name__ == '__main__':
    main()

配合 FastAPI 发送任务

前端通过 FastAPI 暴露一个 HTTP 接口把任务丢进 SQS，Worker 在另一个容器/进程中异步消费：

python

# fastapi_app.py
from fastapi import FastAPI, HTTPException
import boto3

app = FastAPI()
sqs = boto3.client('sqs', region_name='us-east-1')
QUEUE_URL = "https://sqs.us-east-1.amazonaws.com/123456789/agent-tasks-standard"

@app.post("/agent/query")
async def submit_rag_query(thread_id: str, question: str):
    resp = sqs.send_message(
        QueueUrl=QUEUE_URL,
        MessageBody=json.dumps({"task": "rag_query", "thread_id": thread_id, "question": question}),
        MessageAttributes={
            "thread_id": {"DataType": "String", "StringValue": thread_id},
        },
    )
    return {"message_id": resp["MessageId"], "status": "queued"}

@app.get("/agent/result/{thread_id}")
async def get_result(thread_id: str):
    dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
    item = dynamodb.Table('agent-checkpoints').get_item(Key={'pk': f'thread#{thread_id}', 'sk': 'result'})
    if 'Item' not in item:
        raise HTTPException(status_code=404, detail="结果尚未就绪")
    return item['Item']['data']

🔗 部署细节（Dockerfile、多容器编排、进程管理）参见 FastAPI ch22。

本章小结

Standard 队列适合高吞吐、顺序不敏感的任务；FIFO 队列适合需要严格顺序的场景，以吞吐量换取确定性；
长轮询（WaitTimeSeconds=20）是 SQS 消费者的标配，减少无效请求；
**死信队列（DLQ）**必须为所有生产队列配置，maxReceiveCount 控制在 3–5 之间；
DynamoDB 作为 Agent 状态存储时，用 pk=thread#xxx + 降序 Query 取最新检查点是常用模式；
**条件写入（ConditionExpression）**是处理并发更新的唯一安全方式，不要用"先 get 再 put"的两步模式；
生产级架构通常是：HTTP 接口（FastAPI）→ SQS → 后台 Worker（DynamoDB 持久化结果）→ 前端轮询或 WebSocket 推送。

🛠️ 动手实践

在本地用 LocalStack（docker run -p 4566:4566 localstack/localstack）搭建一个假的 SQS + DynamoDB 环境，编写完整可运行的 agent_worker.py，验证消息从发送→消费→DLQ 的完整链路。
给上面的 Worker 加一个 retry_with_backoff 装饰器：处理失败时等待 2^attempt 秒再重新入队，最多重试 3 次，超过 3 次自动移入 DLQ（通过调 change_message_visibility 把 VisibilityTimeout 设成 0 实现）。
在 DynamoDB 里实现"检查点历史"功能：每次保存 checkpoint 时同时写入 history#0、history#1……，然后用 Query + begins_with(sk, 'history#') 查询某个 thread 的完整执行历史。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. SQS FIFO 队列相比 Standard 队列的核心区别是什么？

2. SQS 消息处理失败时，以下哪种做法是正确且推荐的？

3. DynamoDB 条件写入 ConditionExpression 的主要作用是什么？

4. 在生产 SQS 消费者中，WaitTimeSeconds 设为 20 的目的是什么？

第 17 章 · SQS 消息队列与 DynamoDB ​

17.1 SQS 标准队列与 FIFO 队列 ​

用 boto3 创建队列 ​

17.2 发送与接收消息 ​

发送消息 ​

接收消息（轮询） ​

处理完毕后删除消息 ​

17.3 死信队列（Dead Letter Queue） ​

17.4 DynamoDB：Agent 状态存储 ​

建表 ​

写入检查点（PutItem） ​

读取最新检查点（Query） ​

条件更新（乐观锁） ​

17.5 完整示例：SQS 驱动 + DynamoDB 持久化的 Agent Worker ​

配合 FastAPI 发送任务 ​

本章小结 ​

🛠️ 动手实践 ​

🧪 随堂测验

第 17 章 · SQS 消息队列与 DynamoDB

17.1 SQS 标准队列与 FIFO 队列

用 boto3 创建队列

17.2 发送与接收消息

发送消息

接收消息（轮询）

处理完毕后删除消息

17.3 死信队列（Dead Letter Queue）

17.4 DynamoDB：Agent 状态存储

建表

写入检查点（PutItem）

读取最新检查点（Query）

条件更新（乐观锁）

17.5 完整示例：SQS 驱动 + DynamoDB 持久化的 Agent Worker

配合 FastAPI 发送任务

本章小结

🛠️ 动手实践