企业官网建设流程全解析

先说说最近看到的一个有意思的项目。有人在 Hacker News 上展示了一个 AI 助手，卖点不是聊天对话多顺畅，而是它自带一个独立收件箱（inbox）。用户可以往里面投递任务，助手异步消费、处理、回填结果，整个流程像一套轻量级的消息队列。这种设计在 AI Agent 工程实践里越来越常见：当助手不再只是“问答机器人”，而需要处理批量写作、定时巡检、工单分类、内容审核等异步任务时，同步聊天的模式就不够用了。

本文会从零实现一个“带收件箱的 AI 助手”后端服务，只依赖 FastAPI 和 Python 标准库。我们会逐步拆解收件箱的任务模型、状态流转、Worker 消费逻辑，以及完整的 API 接口，最后给出常见问题排查思路和生产环境落地的建议。无论你是刚开始接触 AI 工程化，还是想给已有助手系统增加异步任务能力，这篇教程都值得收藏。

1. 背景与核心概念

1.1 什么是“带收件箱的 AI 助手”

传统的 AI 助手通常是一个同步聊天接口：用户发送问题，模型返回回答，调用结束后整个交互就结束了。这个模式对“聊天”场景没问题，但对任务型场景存在明显短板。比如用户一次性提交 20 封邮件让助手生成摘要，如果同步处理，客户端必须长时间等待，一旦网络波动或接口超时，前面所有结果都丢失了。

“带收件箱的 AI 助手”借鉴了异步消息系统的设计思路。助手内部维护一个收件箱，所有请求先进入收件箱排队，后台 Worker 不断从收件箱中取出任务，调用模型或工具处理，再把结果回写到对应的任务记录上。用户提交请求后拿到一个task_id，之后可以通过这个 ID 查询处理进度和最终结果。

从架构角度看，收件箱本质上是一个任务队列，只是它需要额外支持：

• 任务状态管理（排队中、处理中、已完成、失败）。
• 按优先级或时间排序取数。
• 任务结果回写与查询。
• 失败重试与错误信息记录。

这套设计并不新鲜，消息队列领域已经实践了多年。但当它被应用到 AI 助手场景时，有一个很大的区别：AI 模型调用往往是慢操作，而且可能失败，必须把“任务状态”和“处理结果”作为一等公民来管理。

1.2 收件箱模式解决的核心问题

给 AI 助手引入收件箱模式，主要解决四个问题。

第一是解耦。调用方只需要把任务投递到收件箱，不需要关心 AI 模型什么时候处理完。后端可以随时增加消费 Worker，也可以平滑升级模型服务，调用方是无感的。

第二是可靠性。任务被持久化后，即使 Worker 进程崩溃，任务记录不会丢失。重启后可以继续消费未完成的任务，这比同步调用里的“请求丢失”场景可靠得多。

第三是可观测性。所有任务都有明确状态，我们可以方便地统计队列积压量、平均处理时长、失败率，甚至对每个任务做审计。

第四是并发可控。我们可以限制同时处理的任务数量，避免大批量请求瞬间压垮模型 API，也可以结合令牌桶做限流。

下面用一个对比表来总结同步聊天与收件箱模式的区别：

1.3 典型应用场景

在实际项目中，这种模式很适合以下场景。

• 内容生成与摘要：批量生成商品文案、新闻摘要、邮件回复草稿。
• 工单分类与回复：客服工单进入收件箱，AI 自动打标、分配、生成建议回复。
• 数据处理任务：从数据库或文件中抽取数据，交给模型结构化，再写回存储。
• 定时巡检报告：每天定时把运营数据丢进收件箱，模型生成日报后推送通知。

这些场景的共同点是：任务到达时间和处理时间不一定是同步的，而且单次处理可能耗时几十秒甚至几分钟。用收件箱模式能最大程度降低系统耦合度。

2. 系统架构与消息状态设计

2.1 整体架构

我们设计的系统包含四个核心角色：API 层、Inbox 存储层、Worker 消费层、AI 处理服务。

下面用一张 ASCII 架构图表示数据流：

客户端 (提交任务/查询结果) │ ▼ ┌─────────────────┐ │ FastAPI 层 │ /inbox/tasks 提交 │ │ /inbox/tasks 查询 └─────────────────┘ │ ▼ ┌─────────────────┐ │ Inbox 存储 │ 任务状态 + 内存队列 └─────────────────┘ │ │ 领取待处理任务 ▼ ┌─────────────────┐ │ AI Worker │ 多线程 / 多进程消费 └─────────────────┘ │ │ 调用模型 / 工具 ▼ ┌─────────────────┐ │ AI 服务 │ LLM API / 本地模型 / 脚本 └─────────────────┘ │ └── 处理完成 → 结果回写 Inbox → 客户端查询

API 层负责接收用户请求，把任务写入 Inbox。Inbox 存储任务记录和状态，Worker 定期从中领取任务。领取后，Worker 调用 AI 处理服务，最后把结果回写到对应任务记录上。

线程模型上，我们的示例采用「API 线程 + 后台 Worker 线程」的方式。FastAPI 启动时拉起一个后台 Worker 线程，Worker 轮询 Inbox，每次领取一个任务。生产环境可以把这个模型替换成多进程 Worker 或独立部署的任务消费者。

2.2 任务生命周期

任务在收件箱中会经历多个状态。这里把状态定义清楚，是整个系统设计的核心。

• PENDING：任务已进入收件箱，等待 Worker 领取。
• PROCESSING：任务被某个 Worker 领取，正在调用 AI 处理。
• DONE：处理成功，结果字段已回填。
• FAILED：处理多次重试仍然失败，错误信息已记录。

状态流转可以用下面一段伪代码表示：

提交任务 --> PENDING Worker 领取 --> PROCESSING 处理成功 --> DONE 处理失败且还有重试次数 --> PENDING 处理失败且达到最大次数 --> FAILED

这里把“失败后重试”和“失败最终态”区分开，非常关键。AI 模型接口经常因为网络抖动、限流、内容审核等原因失败，如果一律进入 FAILED，会让很多本来可以成功的任务白白失败；如果无限重试，又会造成成本浪费和队列堆积。常见方案是设置最大尝试次数，比如 3 次，前 2 次失败回到 PENDING，第 3 次失败进入 FAILED。

3. 环境准备与项目结构

3.1 运行环境与依赖

本文示例代码使用 Python 3.10+，主要依赖 FastAPI 和 Uvicorn。数据库方面先用内存存储演示，后续可以在最佳实践章节替换为 Redis 或 SQLite。

你需要准备的环境如下：

• Python 3.10 或更高版本。
• 一个虚拟环境（venv 或 conda 均可）。
• pip 安装 fastapi、uvicorn、pydantic。

版本不需要刻意固定，本文示例以常见环境为准，重点演示设计思路。下面代码基于 pydantic v2 编写；如果你使用的是 pydantic v1，需要把model_dump(mode="json")改回dict()，或者直接在模型中写自定义序列化方法。

3.2 项目目录结构

为了方便阅读，我们将代码拆成几个模块，结构如下：

ai-inbox-assistant/ ├── requirements.txt ├── app/ │ ├── __init__.py │ ├── models.py │ ├── inbox.py │ ├── worker.py │ └── main.py └── README.md

各个文件职责如下：

4. 核心模块设计详解

4.1 消息模型定义

先定义收件箱中的任务模型。它比普通队列消息多了一些业务字段：发送方、主题、优先级、内容、状态、尝试次数、结果和错误信息。

字段设计说明：

• id：任务唯一标识，用 UUID 生成。
• sender：消息来源，比如email、user、cron。
• topic：任务主题或分类，方便后续筛选。
• content：要交给 AI 处理的原始内容。
• priority：优先级，影响 Worker 取数的先后顺序。
• status：任务当前状态。
• attempts：当前已尝试执行次数，用于失败重试。
• result：处理成功后的结果。
• error：最近一次失败的错误信息。

优先级建议使用枚举，这样在 API 层校验参数时更安全。我们定义TaskPriority枚举，包含HIGH、NORMAL、LOW三档。任务状态用TaskStatus枚举，包含PENDING、PROCESSING、DONE、FAILED四种。

4.2 Inbox 存储与取数策略

这里我们用 Python 内置字典作为任务存储，用RLock保证线程安全。收件箱需要提供以下能力：

• add：添加任务，返回新任务对象。
• claim_next：领取下一个待处理任务。
• complete：完成任务，回填结果。
• fail_or_retry：处理失败，判断是否重试或进入最终失败态。
• get：按 ID 查询任务。
• list：列出任务，支持按状态筛选。

claim_next是核心方法。Worker 调用它时，必须保证“找出任务”和“修改状态为 PROCESSING”是原子的，否则多个 Worker 同时消费时会拿到同一个任务，造成重复处理。这里我们在锁内完成查找和状态更新，保证了单进程内多个线程不会重复领取。

取数策略上，我们支持按优先级排序。同一优先级的任务按创建时间先后处理，这样既满足业务紧急度要求，又不会让低优先级任务无限积压。

4.3 Worker 消费端

Worker 是一个后台线程，循环执行以下步骤：

1. 调用claim_next()领取任务。
2. 如果当前没有任务，休眠 1 秒再继续。
3. 调用 AI 处理逻辑。
4. 成功则调用complete()回填结果。
5. 失败则调用fail_or_retry()记录错误并决定是否重试。

Worker 使用daemon线程的原因是不阻塞主进程退出。在真实生产环境中，建议用进程管理工具或容器编排来管理多个 Worker，而不是单线程。

4.4 AI 处理服务抽象

为了演示，我们把“AI 处理”抽象成handle()方法。真实项目中，这个方法内部可以调用 OpenAI 等大模型 API，也可以调用本地部署的模型服务，还可以执行一段工具脚本。

这里有一个设计要点：AI 调用一定要设置超时。模型接口的响应时间往往不稳定，如果 Worker 因为没有超时而卡在一个任务上，后续所有任务都会被阻塞。我们可以在handle()中显式设置 HTTP 客户端超时，或者用Signal强制中断同步调用。

5. 完整实现：FastAPI + 线程 Worker

5.1 创建项目与安装依赖

首先创建项目目录和虚拟环境。

mkdir ai-inbox-assistant cd ai-inbox-assistant python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate

创建requirements.txt并写入以下依赖：

fastapi>=0.110 uvicorn[standard]>=0.29 pydantic>=2.0

安装依赖：

pip install -r requirements.txt

5.2 定义数据模型

文件路径：app/models.py

from datetime import datetime, timezone from enum import Enum from typing import Optional from pydantic import BaseModel, Field def utc_now() -> datetime: """统一获取当前 UTC 时间，避免重复实现。""" return datetime.now(timezone.utc) class TaskStatus(str, Enum): PENDING = "pending" PROCESSING = "processing" DONE = "done" FAILED = "failed" class TaskPriority(str, Enum): HIGH = "high" NORMAL = "normal" LOW = "low" class InboxTask(BaseModel): id: str = Field(default_factory=lambda: __import__("uuid").uuid4().hex) sender: str = "unknown" topic: str = "default" content: str priority: TaskPriority = TaskPriority.NORMAL status: TaskStatus = TaskStatus.PENDING created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) attempts: int = 0 result: Optional[str] = None error: Optional[str] = None def to_dict(self) -> dict: """转为可直接 JSON 序列化的字典，兼容 pydantic v1/v2。""" return { "id": self.id, "sender": self.sender, "topic": self.topic, "content": self.content, "priority": self.priority.value, "status": self.status.value, "created_at": self.created_at.isoformat(), "updated_at": self.updated_at.isoformat(), "attempts": self.attempts, "result": self.result, "error": self.error, }

这里重点解释几个设计决策。

id使用uuid4().hex生成 32 位十六进制字符串，足以避免并发提交时的 ID 冲突。也可以直接用str(uuid.uuid4())，区别只是是否带横线。

to_dict()方法统一负责序列化，把枚举值、时间对象转换为普通字符串，这样接口层在返回响应时不需要关心底层 pydantic 版本差异。attempts字段默认 0，表示任务还未被消费。

5.3 实现 Inbox 核心逻辑

文件路径：app/inbox.py

import threading from typing import Dict, List, Optional from .models import InboxTask, TaskPriority, TaskStatus class Inbox: """线程安全的内存收件箱。""" def __init__(self) -> None: self._tasks: Dict[str, InboxTask] = {} self._lock = threading.RLock() @staticmethod def _sort_key(task: InboxTask): """优先级高的任务排在前面，相同优先级按创建时间判断。""" priority_order = { TaskPriority.HIGH: 0, TaskPriority.NORMAL: 1, TaskPriority.LOW: 2, } return (priority_order.get(task.priority, 1), task.created_at) def add(self, content: str, sender: str = "unknown", topic: str = "default", priority: TaskPriority = TaskPriority.NORMAL) -> InboxTask: """向收件箱添加一个任务。""" with self._lock: task = InboxTask( sender=sender, topic=topic, content=content, priority=priority, ) self._tasks[task.id] = task return task def claim_next(self) -> Optional[InboxTask]: """领取下一个待处理任务，并将状态改为 PROCESSING。""" with self._lock: candidates = [ task for task in self._tasks.values() if task.status == TaskStatus.PENDING ] if not candidates: return None candidates.sort(key=self._sort_key) task = candidates[0] task.status = TaskStatus.PROCESSING task.attempts += 1 task.updated_at = __import__("app.models", fromlist=["utc_now"]).utc_now() return task def complete(self, task_id: str, result: str) -> None: """处理成功后回填结果。""" with self._lock: task = self._tasks.get(task_id) if task is None: raise KeyError(f"task {task_id} not found") task.status = TaskStatus.DONE task.result = result task.error = None task.updated_at = __import__("app.models", fromlist=["utc_now"]).utc_now() def fail_or_retry(self, task_id: str, error: str, max_attempts: int = 3) -> None: """错误处理：如果未超过最大执行次数则回到 PENDING，否则标记 FAILED。""" with self._lock: task = self._tasks.get(task_id) if task is None: raise KeyError(f"task {task_id} not found") task.error = error task.updated_at = __import__("app.models", fromlist=["utc_now"]).utc_now() if task.attempts < max_attempts: task.status = TaskStatus.PENDING else: task.status = TaskStatus.FAILED def get(self, task_id: str) -> Optional[InboxTask]: with self._lock: return self._tasks.get(task_id) def list(self, status: Optional[TaskStatus] = None) -> List[InboxTask]: with self._lock: tasks = list(self._tasks.values()) if status is not None: tasks = [t for t in tasks if t.status == status] tasks.sort(key=lambda t: t.created_at) return tasks

这里有一个实现细节需要注意：claim_next返回的是任务对象本身，而不是副本。这意味着 Worker 在拿到任务对象后，即使 Inbox 锁已经释放，其他线程读取这个任务时也能看到PROCESSING状态。这正是我们希望的效果，因为它反映了真实的执行状态。

不过要注意，由于我们直接修改任务对象的属性，如果 Worker 在执行任务时不小心修改了content等业务字段，会产生脏数据。所以在设计约定上，Worker 只允许通过complete和fail_or_retry修改任务状态，不要直接操作字段。

5.4 实现 Worker 消费端

文件路径：app/worker.py

import threading import time from typing import Optional from .inbox import Inbox from .models import InboxTask class AIWorker(threading.Thread): """后台消费线程：从 Inbox 领取任务、调用模型、回填结果。""" def __init__(self, inbox: Inbox, name: str = "ai-worker", poll_interval: float = 1.0): super().__init__(name=name, daemon=True) self.inbox = inbox self.poll_interval = poll_interval self._stop_event = threading.Event() def stop(self) -> None: self._stop_event.set() def run(self) -> None: while not self._stop_event.is_set(): task: Optional[InboxTask] = self.inbox.claim_next() if task is None: self._stop_event.wait(self.poll_interval) continue try: result = self.handle(task) self.inbox.complete(task.id, result) except Exception as exc: self.inbox.fail_or_retry(task.id, str(exc)) def handle(self, task: InboxTask) -> str: """核心 AI 处理函数，可替换为真实模型 API 调用。""" # 模拟耗时操作，生产环境替换成 LLM API / 本地模型推理 time.sleep(0.5) return f"[{task.topic}] {task.content[:20]} 的 AI 摘要已生成"

Worker 中最容易踩坑的是异常处理边界。handle()中任何异常都会触发fail_or_retry()，但这个逻辑需要与重试策略配合。如果任务是“永久性错误”，例如内容包含非法字符导致模型拒绝处理，重试多少次都不成功，反而会浪费资源。所以在生产系统中，handle()内部应该区分临时错误和永久错误，永久错误直接抛出特定异常，由调用方判断是一次性失败还是继续重试。

这个示例中的poll_interval是 1 秒，在演示环境可以接受。生产环境通常用消息队列的阻塞读取或者长轮询，避免无意义的轮询开销。

5.5 编写 FastAPI 接口

文件路径：app/main.py

from contextlib import asynccontextmanager from typing import Optional from fastapi import FastAPI, HTTPException, Query from .inbox import Inbox from .models import InboxTask, TaskStatus from .worker import AIWorker inbox = Inbox() worker: Optional[AIWorker] = None @asynccontextmanager async def lifespan(app: FastAPI): global worker worker = AIWorker(inbox, name="ai-worker") worker.start() yield if worker is not None: worker.stop() app = FastAPI( title="AI Assistant with Inbox", description="一个自带收件箱的 AI 助手服务", version="0.1.0", lifespan=lifespan, ) class TaskCreateRequest: def __init__(self, content: str, sender: str = "unknown", topic: str = "default", priority: str = "normal"): self.content = content self.sender = sender self.topic = topic self.priority = priority from pydantic import BaseModel class TaskCreateBody(BaseModel): content: str sender: str = "unknown" topic: str = "default" priority: str = "normal" class TaskListResponse(BaseModel): items: list[dict] @app.post("/inbox/tasks", status_code=201) def create_task(body: TaskCreateBody) -> dict: """提交一个新任务到收件箱。""" from .models import TaskPriority try: priority = TaskPriority(body.priority) except ValueError: raise HTTPException(status_code=422, detail=f"无效优先级: {body.priority}") task = inbox.add( content=body.content, sender=body.sender, topic=body.topic, priority=priority, ) return {"task_id": task.id, "status": task.status.value} @app.get("/inbox/tasks") def list_tasks( status: Optional[TaskStatus] = Query(default=None), sender: Optional[str] = Query(default=None), ) -> TaskListResponse: """列出收件箱任务，支持按状态和发送方筛选。""" tasks = inbox.list(status=status) if sender: tasks = [t for t in tasks if t.sender == sender] return TaskListResponse(items=[t.to_dict() for t in tasks]) @app.get("/inbox/tasks/{task_id}") def get_task(task_id: str) -> dict: """查询单个任务状态和结果。""" task: Optional[InboxTask] = inbox.get(task_id) if task is None: raise HTTPException(status_code=404, detail="任务不存在") return task.to_dict()

代码里保留了TaskCreateRequest这个旧类，其实是不需要的，可以去掉。我在这里故意保留是因为实际开发中经常会有“写多了再清理”的情况。正式代码建议直接删掉，只保留 Pydantic 模型。

接口设计有三个核心点。

第一，POST /inbox/tasks返回task_id而不是完整处理结果。客户端拿到任务 ID 后，可以通过GET /inbox/tasks/{task_id}轮询结果。这是异步任务接口的标准做法。

第二，查询接口支持按status和sender过滤，方便业务侧按状态或来源查看收件箱内容。

第三，状态枚举通过 Query 参数接收时，FastAPI 会自动做参数校验。如果传入非法状态，返回 422，不需要我们手写校验逻辑。

5.6 启动服务并验证

现在启动服务。

uvicorn app.main:app --reload --port 8000

看到如下输出说明启动成功：

INFO: Uvicorn running on http://127.0.0.1:8000 INFO: Application startup complete.

FastAPI 会自动生成交互式文档，访问http://127.0.0.1:8000/docs可以查看所有接口。

6. 运行演示与结果说明

6.1 提交任务

打开另一个终端，使用 curl 提交两个测试任务，一个高优先级，一个普通优先级。

curl -X POST http://127.0.0.1:8000/inbox/tasks \ -H "Content-Type: application/json" \ -d '{"content": "请总结本周运营数据", "sender": "cron", "topic": "report", "priority": "high"}'

预期输出：

{"task_id":"9f7b2f6d0c9a4e6f9c48e0c9ae62da21","status":"pending"}

再提交一个普通任务：

curl -X POST http://127.0.0.1:8000/inbox/tasks \ -H "Content-Type: application/json" \ -d '{"content": "生成一封客户回复邮件", "sender": "user", "topic": "email", "priority": "normal"}'

6.2 查询任务列表

任务提交后，立即查询列表，可能看到部分任务处于pending，部分处于processing，取决于 Worker 的处理速度。

curl http://127.0.0.1:8000/inbox/tasks

输出示例：

{ "items": [ { "id": "9f7b2f6d0c9a4e6f9c48e0c9ae62da21", "sender": "cron", "topic": "report", "content": "请总结本周运营数据", "priority": "high", "status": "done", "created_at": "2025-01-01T10:00:00+00:00", "updated_at": "2025-01-01T10:00:01+00:00", "attempts": 1, "result": "[report] 请总结本周运营数据 的 AI 摘要已生成", "error": null } ] }

注意attempts字段已经变成 1，说明 Worker 领取并处理过一次。result字段已经回填生成结果。

6.3 查询单个任务结果

根据之前拿到的task_id，查询单个任务：

curl http://127.0.0.1:8000/inbox/tasks/9f7b2f6d0c9a4e6f9c48e0c9ae62da21

输出与列表中的单个项目一致。到这里，一个最小的“带收件箱的 AI 助手”已经可以跑通了。

7. 常见问题与排查思路

实际开发中，你会遇到各种预期外的情况。下面整理了一些高频问题。

7.1 任务一直处于 pending 状态

出现这个现象，首先检查 Worker 是否在运行。在启动日志中看不到 Worker 相关信息时，往往是 lifespan 生命周期没有挂载正确。

FastAPI 旧版本常见做法是@app.on_event("startup")，新版开始推荐lifespan上下文管理器。如果你使用的是较老版本 FastAPI，可以改回 startup 事件写法，但要注意不同版本的兼容性。

还可以在 Worker 的run()方法最开始加一行打印日志，比如print("[worker] started")，这样能很快确认线程是否启动。

7.2 任务重复消费

在单进程多线程模型中，claim_next因为有RLock保护，不会出现重复领取。但在多进程部署时，每个进程都有自己的 Inbox 实例，任务存储不共享，这时候问题会变成“各进程各处理各的”，而不是重复消费同一个任务。

真正的重复消费风险发生在：任务存储是共享的（比如 Redis），但领取时没有用原子操作。解决方法有两种：

• 在 Inbox 存储层使用带条件的原子更新，例如 Redis Lua 脚本或 SQLUPDATE ... WHERE status='pending'。
• 在 Worker 处理结果回写时，使用幂等 ID 校验，防止重复写入结果。

对于 AI 任务，重复消费不只是资源浪费，还可能导致重复扣费和重复生成内容，所以幂等设计要提前做。

7.3 模型调用超时

模型 API 是外部依赖，它的延迟不可控。如果不设置超时，一个慢请求可能让 Worker 长期阻塞。常见做法有：

• 在网络请求库层面设置timeout，比如requests.post(url, timeout=(3, 30))。
• 在多线程 Worker 中，用Future.get(timeout=...)控制单个任务执行时长。
• 为任务设置最大执行时间，超过阈值的任务重新进入队列或直接标记失败。

8. 最佳实践与工程建议

演示代码跑通后，如果要在生产环境落地，下面这些点非常关键。

8.1 存储层选型

内存字典最明显的缺点是重启丢数据。生产环境推荐替换为以下方案之一。

如果你已经有 RabbitMQ 或 Kafka 基础设施，也可以直接把它们作为任务队列，但要在消息体里保留task_id和完整错误信息。

8.2 幂等与重试策略

AI 调用通常涉及成本，重试策略必须谨慎。

建议按以下原则设计：

• 为每个任务生成全局唯一request_id，发往模型服务时携带该 ID。
• 网络超时、限流、5xx 等临时错误允许重试。
• 内容不合法、参数错误等永久错误不要重试。
• 设置最大尝试次数，默认为 3，避免无限重试。
• 使用指数退避策略，比如第 1 次等 2 秒，第 2 次等 4 秒，第 3 次等 8 秒。

在当前的fail_or_retry方法中，最简单的指数退避可以放在 Worker 内部实现：重试前time.sleep(backoff)。

8.3 超时与死信

任务长时间处于PROCESSING状态，可能是 Worker 崩溃导致的任务“死亡”。生产环境需要引入“死信”机制。

可以每隔一段时间扫描状态为PROCESSING但updated_at超过 10 分钟的任务，将它们重新置为PENDING或标记为FAILED，并记录告警。这个扫描任务通常由定时调度器执行。

8.4 安全与鉴权

收件箱中可能包含敏感数据，比如客户邮件、业务报告文本。接口不能裸奔在公网上。

建议：

• 在 FastAPI 中配置 API Key 或 OAuth2 鉴权。
• 对任务内容加密存储。
• 查询接口做权限校验，普通用户只能查询自己提交的任务，不能查看他人的任务内容。
• 记录每个请求的操作人、时间和任务 ID，以便审计。

8.5 AI 调用成本控制

当收件箱堆积大量任务时，如果不做控制，模型 API 账单会很快飙升。

控制成本可以从几个方向入手：

• 任务入库前进行内容长度限制和去重。
• 对相同或近似内容做缓存，命中后直接返回历史结果。
• 给 Worker 加速率限制，防止瞬间请求过多导致模型 API 限流。
• 流式读取大文本时先做预处理，