如何设计 Agentic 系统 2 ：Agent 后端系统的八层设计与实践

3 部分中的第 2 部分，agentic 应用的参考架构：API、队列、workers、流式输出、结构化输出、会话，以及阻止 bug 变成账单的限制。

在此阅读关于如何设计 Agentic 系统的第 1 部分。

在一切其他方案生效之前，先扼杀这个本能：

@app.post("/agent")
def run(req: Request):
    return agent.run(req.prompt)     # 90 秒后，也许

这是最自然的写法，但它在六个方面都是错误的，每一个都会在不同的一天变成一次事故。它让连接保持打开两分钟。标签页关闭时任务就丢失了。网络抖动时它会重试一个 $0.20 的操作。它给用户一个加载动画而没有任何信息。它无法被取消。当它在第 97 秒失败时，前 96 秒都白费了。第 1 部分涵盖了 harness：工具、提示词、记忆、编排、评估。这是运行它的后端：一个可以拿来即用并改造的模板。示例使用 FastAPI，因为这是最常见的情况，但这里没有任何内容依赖它。第 3 部分将其部署到 AWS 上。

下面的一切都源于工作负载的四个属性：

形态

八个层加一个控制平面。每个 agent 后端都有这些，无论是否有意设计。

控制平面——即限制、配额、追踪——不是第九个盒子；它是同样的少数几个关注点，在八个层中的每一层内部反复出现。 这样布局项目，让各层清晰可见：

app/
├── main.py            # 应用工厂、中间件、路由注册
├── config.py          # 从环境变量读取设置——单一来源，不散落 os.getenv
├── deps.py            # 依赖：租户、数据库会话、速率限制
├── schemas.py         # Pydantic 请求/响应模型
├── routes/            # 仅 HTTP：校验、调用服务、返回
│   ├── jobs.py
│   └── health.py
├── services/          # 业务逻辑——这里不导入 FastAPI
├── agent/             # harness（第 1 部分）：工具、提示词、编排
├── models_layer.py    # 提供商接缝——唯一导入 SDK 的文件
├── workers/           # 队列任务
└── store/             # 热存储（Redis）+ 持久存储（Postgres）

两条规则让这个结构保持真实。路由不包含逻辑：它们校验、调用服务、返回。

只有一个文件导入提供商 SDK。如果 import openai 出现在三个地方，你就没有接缝，第 5 层以后就无法添加了。

Layer1：身份与隔离

先做这个；下面的每一层都以它为关键。"每个租户"在请求携带经过验证的主体之前毫无意义。

## deps.py
async def current_tenant(authorization: str = Header(...)) -> Tenant:
    return await verify_token(authorization)   # JWT、API key 或 mTLS

Tenant = Annotated[Tenant, Depends(current_tenant)]

三条规则，现在执行成本低，以后执行成本高：

1. 按经过验证的主体来限定范围，绝不使用传入的主体。 GET /jobs/{id} 按调用者的租户过滤，而不仅仅是 ID。这是最常见的 agent 后端漏洞，因为 job ID 会被记录、分享并粘贴到 URL 中。
2. 让 job ID 不可猜测（UUID4）。 顺序 ID 加上缺失的租户检查就是数据泄露。
3. 将租户带到每一层： 队列消息、worker 上下文、缓存键、日志行、追踪、预算。没有租户作为键的缓存最终会把一个客户的答案提供给另一个客户：在测试中不可见，在公开环境中无法恢复。

Layer2：API 表面

三个动词。不是一个。

Submit 在工作开始之前就返回。这个决定换来了超时独立性、标签页关闭后的存活、独立扩缩容，以及清晰的重试边界。

POST   /jobs                → 202 { job_id, status: "queued" }
GET    /jobs/{id}           → { status, step, result? }
GET    /jobs/{id}/events    → SSE 流（第 8 层）
POST   /jobs/{id}/cancel    → 202
POST   /jobs/{id}/resume    → 202   （批准 / 拒绝 / 回答）

Pydantic 是你的第一道防线

这里的校验不是形式主义，它是阻止坏请求变成账单的最便宜的地方。

class JobRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")     # 拒绝未知字段

    prompt: str = Field(min_length=1, max_length=50_000)
    documents: list[str] = Field(default_factory=list, max_length=20)
    temperature: float = Field(default=0.3, ge=0.0, le=1.0)

    @field_validator("documents")
    @classmethod
    def total_size(cls, v):
        if sum(len(d) for d in v) > 200_000:
            raise ValueError("documents exceed 200KB total")
        return v

extra="forbid" 能捕获拼写错误的字段名，否则它们会被静默忽略。边界才是真正的重点：一个 100KB 的文档是合法的 HTTP body，却是一个毁灭性的提示词。

Submit 做五件事，没有一件是运行 agent

@router.post("/jobs", status_code=202, response_model=JobHandle)
async def submit(
    req: JobRequest,
    tenant: Tenant,
    idem_key: str | None = Header(None, alias="Idempotency-Key"),
):
    if existing := await idem.lookup(tenant.id, idem_key):   # 1. 幂等性
        return existing
    est = estimate_tokens(req)                                # 2. 成本上限
    if est > MAX_INPUT_TOKENS:
        raise HTTPException(413, f"input too large: ~{est} tokens")
    job = await store.create(tenant.id, req, status="queued") # 3. 先持久化
    await queue.enqueue(job.id)                               # 4. 然后入队
    return JobHandle(job_id=job.id, status="queued")          # 5. 立即返回

幂等性在这里不是可选项。在普通 API 中，重复提交只是不整洁；在 agent API 中，它要花钱。一次双击、一次客户端重试或一个后台移动请求都会产生两个 job 和两张账单。使用标准的 Idempotency-Key header：任何不透明的客户端字符串，存储 24 小时，重复提交时返回原始 handle。

先持久化再入队。反过来，worker 可能会捡起 store 不知道的 job，于是 GET /jobs/{id} 对一个正在运行的 job 返回 404。

把状态机写下来

queued → running → [step_1 … step_n] → completed
   │        │              │
   │        │              └→ awaiting_input ──(恢复)──┘
   │        └→ failed
   └→ cancelled

六个状态，三个终态。人工审批是其中之一，不是一个标志位。

中间状态就是整个用户体验。一个看着 "running" 90 秒的用户会认为它坏了；一个看着 extracting → drafting → reviewing 的用户会等待。成本：每次状态转换一次热存储写入。

awaiting_input 是一等状态，不是一个标志位——human-in-the-loop 意味着保持一个 job 半完成状态，并从与暂停它的不同进程恢复它。给它一个过期时间；一个七天没有人批准的 job 应该过期，而不是永远等待。

取消需要一个机制，而不只是一个 URL

这是你拥有的最便宜的成本控制手段，而且通常是一个死端点。

## API：设置一个 worker 能看到的标志
await hot.set(f"cancel:{job_id}", 1, ex=3600)

## Worker：在步骤之间检查，并中止进行中的生成
if await hot.exists(f"cancel:{job_id}"):
    await stream.aclose()          # 立即停止为 token 付费
    raise Cancelled(job_id)

如果你的步骤很长，将取消Token传入模型层，这样进行中的流可以在生成中途被中止，而不是等到下一个步骤边界。

第 3 层：队列

承重墙：超时独立性、独立扩缩容、优雅降级而非失败的背压，以及明确的重试边界。

至少三个队列：默认队列、优先队列（交互式或付费层级，使用独立 workers），以及一个由人工检查且绝不自动清空的 DLQ。一旦你有两个客户，就添加每租户公平性——一个租户提交 500 个 job 不能饿死其他租户。

承重的 Worker 设置

acks_late             = True    # 完成时确认，而非接收时
prefetch_count        = 1       # 长任务：绝不囤积消息
max_tasks_per_child   = 50      # 回收——agent SDK 会泄漏
soft_time_limit       = 900     # 15 分钟——在任务内部抛出
hard_time_limit       = 1020    # 17 分钟——杀死进程
retry_on_worker_lost  = False   # 挂起的 job 绝不能重新投递

agent 循环没有自然的终点。框架的递归限制计算的是轮次，而不是秒。一个在轮次之间停滞的 job——提供商挂起一个 socket、工具等待连接——会永远运行并永远占用它的槽位。墙上时钟限制是唯一真正的兜底：soft limit 在任务内部抛出，这样你可以记录一个用户能读到的原因；当框架的宽泛 except 吞掉 soft limit 时，hard limit 杀死进程。

retry-on-worker-lost 是昂贵的默认设置。使用 late acks 时，被 hard limit 杀死的任务会被视为"worker 丢失"并被重新投递——所以一个挂起的 job 会在下一个 worker 上再次挂起，再下一个，永远烧钱。让它失败一次就好。

规则：重试传输层，绝不重试推理。重试 429、断开的 socket、503。一个运行了十五分钟且没有产生任何结果的 job 不是瞬时故障。

递归限制计算轮次；停滞的循环停止计数。只有时钟能结束它。

第 4 层：Workers 与并发

大多数团队在这里留下 10 倍的性能空间，因为他们用 Web 服务的直觉来规划 agent worker 的规模。一个 agent worker 将 70–95% 的墙上时钟时间花在阻塞于网络 socket 上。它不是 CPU 密集型；它几乎不感知 CPU。

默认采用 进程 × 异步：每个核心一个进程用于隔离和泄漏遏制，每个进程内部高异步并发。

吞噬异步后端的陷阱：异步路径中的一个同步调用会阻塞整个事件循环——阻塞式 SDK、同步数据库驱动、time.sleep、重型分词器。在负载下，症状是吞吐量无缘无故地崩溃。

##  阻塞此进程中的每个其他请求
@router.post("/x")
async def bad(): return blocking_sdk_call(prompt)

##  要么使用异步客户端，要么把它推出事件循环
@router.post("/x")
async def good(): return await run_in_threadpool(blocking_sdk_call, prompt)

一个同步调用冻结进程中的每个请求；await 让它们交错执行。

用算术而不是感觉来规划规模

concurrency_per_worker = (provider_RPM / 60) × avg_job_duration_s / calls_per_job
workers_needed         = ceil(arrival_rate_per_s × job_duration_s / concurrency_per_worker)

Job 以 0.5/秒到达，耗时 90 秒，每个调用 8 次模型，提供商允许 600 RPM：

提供商上限：$(600/60) \times 90 / 8 = 112$ 个并发 job，系统范围

Little's Law：$0.5 \times 90 = 45$ 个在稳态下进行中

每个 worker 25 并发 → 2 个 workers，在限流之前还有余量

你的上限通常是提供商，而不是你的基础设施。垂直扩展几乎没有帮助——先提高每 worker 并发（免费），然后添加 workers。

内存才是每个 worker 的真正限制：每个进行中的 job 持有对话状态、工具结果和文档；每个 job 20–50MB，并发 100 就是 2–5GB。用信号量限制进行中的模型调用，这样突发不会变成一堵 429 的墙。

优雅关闭，否则每次部署都是一次事故

SIGTERM 在每次部署、缩容和 spot 回收时都会在任务中途到达。对于 90 秒的 job，默认的 30 秒宽限期会杀死进行中的工作。

SIGTERM → 停止消费新消息
        → 让进行中的 job 完成（宽限期 ≥ p99 时长）
        → 在截止时间重新排队仍在运行的内容
        → 退出

将平台的终止宽限期设置为高于 p99 job 时长，并在 SIGTERM 时立即让 readiness 探针失败，同时在排空期间保持 liveness 探针存活。

第 5 层：模型调用层

几乎所有成本和可靠性都在这里。绝不要从 agent 代码中调用提供商 SDK——在它前面放一个接缝，并给它六项工作。

精确缓存、语义缓存、路由器、提供商——由熔断器决定队列是否继续前进。

1. 三层缓存

它们叠加生效，最便宜的在前。提示词缓存是本文中杠杆率最高的改动，而且它主要是一个布局问题——静态在前，动态在后，始终如此：

[ system prompt ][ tool definitions ][ few-shot examples ]  ← 静态，已缓存
[ retrieved docs ][ conversation ][ user input ]            ← 动态，未缓存

把一个变量移到那条线之上——时间戳、用户 ID、打乱的工具列表——命中率就会悄悄归零。没有任何报错；账单只是翻倍。用提示词对缓存键做版本控制，否则一次编辑就会毒化所有条目。

2. 按任务路由

一个五步流水线通常只有一步需要前沿模型，其余四步不需要：提取、分类、路由和格式化交给小模型；草拟和工具推理交给中档模型；最终审查和评判交给前沿模型。按步骤路由，60–80% 的调用会降一个档次，没有可衡量的质量损失，这是继缓存之后最大的杠杆。

同样的五步按档次路由：每次运行 $0.087 而不是 $0.300。

3. 带抖动重试，并对失败分类

RETRYABLE = {429, 500, 502, 503, 504, ConnectionError, Timeout}

for attempt in range(MAX_ATTEMPTS):
    try:
        return await provider.call(...)
    except err if classify(err) in RETRYABLE:
        backoff = min(BASE * 2**attempt, CAP) * random.uniform(0.5, 1.5)
        await asyncio.sleep(retry_after(err) or backoff)
    except err:
        raise          # 400、内容过滤器、上下文长度：绝不重试

遵守 Retry-After，并且始终使用抖动——整个集群的同步重试会把一次短暂的限流变成持续的中断。

4. 熔断

当提供商降级时，拉取更多工作只会让情况更糟。熔断器打开时：停止从队列消费。不是"快速失败"——只是停止。这就是队列的用途。一个前面没有熔断器的队列会把 10 分钟的波动变成一波永久性失败。

5. 跨提供商回退

在熔断器打开时进行故障转移，当 p95 超过阈值时切换到备用提供商。代价是纪律：将抽象保持在提供商功能的交集，并在两者上都进行评估。一个你从未测试过的回退只是装饰。

6. 池化、预热、流式、批处理

每个进程一个客户端并复用——每次调用一个客户端就是每次调用一次 TLS 握手。池大小 ≥ 你的并发上限，否则请求会在 HTTP 客户端内部不可见地排队。在 readiness 探针通过之前预热。即使你不向用户流式输出，也要从提供商流式接收，这样失控的生成可以被中止而不是被付费。批处理离线工作——这些 API 大约便宜 50%。

限制每个 job 的支出，在步骤之间检查：

if job.tokens_used > job.token_budget:
    raise BudgetExceeded(job_id)   # 有边界的失败好过无边界账单

结构化输出：校验，然后修复一次

模型返回文本。你的代码想要对象。绝不要对模型响应执行 json.loads() 然后祈祷。

class Extraction(BaseModel):
    title: str
    tags: list[str] = Field(max_length=5)
    confidence: float = Field(ge=0, le=1)

async def extract(text: str) -> Extraction:
    raw = await model.call(prompt(text), schema=Extraction.model_json_schema())
    try:
        return Extraction.model_validate_json(raw)
    except ValidationError as e:
        # 一次修复尝试：把模型自己的错误交给它
        fixed = await model.call(repair_prompt(raw, str(e)), schema=...)
        return Extraction.model_validate_json(fixed)   # 如果仍然不好，就大声失败

三条规则。当提供商的原生结构化输出模式存在时使用它——约束解码胜过"请返回 JSON"。修复一次，不要循环——无界的修复循环就是无界的账单。让第二次失败成为真正的失败；干净的失败胜过默认值悄悄返回的 confidence=0.0。

相同的 Pydantic 模型校验 API 边界和模型边界——这种对称性就是大部分价值所在。

第 6 层：工具执行

工具是 agent 不再是文本生成器而成为拥有凭据的进程的地方。这一层是一个安全边界。

从工具返回的一切都是数据，绝不是指令。

在与你的凭据相同的进程中执行模型生成的代码，没有安全的方式。

工具赋予 agent 凭据和网络可达性，所以它们需要自己的控制。将出站流量锁定到阻止元数据端点和私有网段的允许列表，并为每个工具设置超时和输出限制，因为挂起会超过上游限制，而巨大的响应会摧毁上下文窗口和账单。

将所有返回内容视为不可信数据，绝不是指令，并将不可逆操作置于人工审批之后。对 $(job_id, step, args)$ 上的副作用做键控，这样重试不会重复发送邮件或重复扣费。通过在工具边界注入密钥来避免密钥出现在提示词中，并在渲染之前净化模型输出。

第 7 层：状态与会话

两个存储，两个问题。

从热存储提供状态，从持久存储提供历史，轮询路径永远不会触及你的数据库。TTL 是一个特性——清理应该是存储的属性，而不是 cron 任务。

能扛过中断的会话

每个聊天形态的 agent 应用在第二周都会遇到的问题：用户刷新、更换设备，或者 worker 在第 80 秒死亡。什么能存活下来？

只存在于进程内存中的东西都无法存活。一个以会话 ID 为键的对话字典在你运行两个 workers 之前完美工作，之后它只有 50% 的时间正常工作。

持久化三样东西，以三种不同的节奏：

## 在每一步之后，而不是每个 token 之后
await store.save_step(job_id, step=step_name, state=agent_state, tokens=used)

## 恢复：重建而不是重启
state = await store.load_step(job_id)
if state:
    agent.resume(from_step=state.step, prior=state.output)

两个设计后果。

以会话为键，而不是以连接为键——会话存在于存储中，任何进程都可以服务它，这正是让在不同设备上重新连接得以工作的原因。

限制你重放的内容：无界增长的会话最终会超过上下文窗口和预算。总结旧的轮次，保留最后 N 轮原文——第 1 部分中的工作/持久记忆划分，以 schema 的形式表达。

三样东西以三种节奏持久化。worker 中的 dict 不在其中。

检查点不是持久执行

检查点（框架级）每步保存状态。它保护应用免受失败：错误的分支、HITL 暂停。但运行存在于一个进程中；如果进程死亡，运行就死亡。数据存活了；执行没有。

持久执行（Temporal、Restate、DBOS）让工作流可以在另一台机器上恢复，从日志中重放已完成的步骤。

现在也决定数据生命周期：每个租户的保留策略、提供商调用前的 PII 编辑、覆盖两个存储以及追踪和缓存的删除路径，以及一个只追加的审计日志。

第 8 层：交付与流式

当进度是粗粒度时轮询——离散状态，没有有意义的中间输出。对热存储每 2 秒轮询一次，在 job 生命周期内大约 50 次廉价读取，并且免费扛过所有网络事件。这不是妥协；这是正确的。

当 token 就是产品时流式输出。SSE 优于 WebSocket：纯 HTTP、单向（用户输入通过普通 POST 返回）、原生重连，而且 MCP 规范转向 Streamable HTTP 使其成为默认选择。

值得指出的错误：直接从 agent 进程流式输出。这重新将连接生命周期与 job 生命周期耦合在一起，并抵消了队列的意义——关闭标签页，丢失工作。在中间缓冲：worker 将事件写入流，无论是否有人监听；API 按需附加。

@router.get("/jobs/{job_id}/events")
async def events(job_id: str, tenant: Tenant, request: Request,
                 last_event_id: str | None = Header(None, alias="Last-Event-ID")):
    await authorize(tenant, job_id)

    async def gen():
        cursor = last_event_id or "0"          # 从他们断开的地方重放
        while not await request.is_disconnected():
            for eid, ev in await hot.xread({f"job:{job_id}:events": cursor},
                                           block=15_000, count=50):
                cursor = eid
                yield f"id: {eid}\nevent: {ev['type']}\ndata: {ev['data']}\n\n"
            yield ": keepalive\n\n"            # 防止代理超时

    return StreamingResponse(gen(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache",
                                      "X-Accel-Buffering": "no"})  # nginx：不要缓冲

三个让初学者吃亏的细节：X-Accel-Buffering: no（否则 nginx 会把你的流缓冲成末尾的一个块，流式看起来就不起作用）、keepalive 注释让空闲连接不被代理清除，以及在 Last-Event-ID 上先重放再续传——这个模式意味着在 token 400 处关闭并在 token 900 处打开的笔记本电脑会看到全部 900 个 token。

SSE 是一种传输，不是存储。即使是完全流式的后端，也保留 GET /jobs/{id} 作为底层的事实来源。

还要处理已经离开的客户端——这是一个用 HMAC 对 body 签名、带时间戳以防重放的 webhook，有自己的重试计划和 DLQ。它也是唯一在没有人工触发 job 时也能工作的交付方式。

衡量流：TTFT 和 tokens/秒

总时长是衡量流式体验的错误指标。两个数字很重要，而且它们独立变化：

t0 = time.perf_counter()
ttft = None; n = 0
async for chunk in stream:
    if ttft is None: ttft = time.perf_counter() - t0
    n += 1
    yield chunk
tps = n / (time.perf_counter() - t0 - ttft)
log.info("model_call", extra={"ttft_ms": ttft*1000, "tps": tps, "tokens": n})

TTFT 是用户能感受到的数字：低于约 500ms 感觉是即时的，超过约 2s 感觉是坏的。它也是你能修复的那个——提示词缓存命中通常能将其减半，而队列等待在提供商指标中不可见，往往是真正的罪魁祸首。TPS 主要取决于模型，但 TPS 崩溃是提供商降级的早期预警。按模型跟踪 p50 和 p95 的这两个指标。

worker 无论是否有监听者都会写入流，所以重连时可以重放错过的内容。

控制平面：限制、扩缩容、可观测性

速率限制以身份为键，而不是 IP——企业 NAT 共享一个 IP，移动客户端在会话中途会更换 IP。计算 token 而不是请求：一个请求可以是 500 个 token 或 500,000 个。网关世界已经收敛到 token 感知配额，以及基于 token 的Token桶，按租户，短窗口应对突发，长窗口应对预算。

返回带 Retry-After 的 429，这样你自己的调用者会正确退避，并豁免健康检查，否则你的负载均衡器会把自己限流出池。

Web 后端限制是为了保护容量。Agent 后端限制是为了保护账单。容量在午夜免费补充；支出不会。

每个闸门都能捕获其他四个闸门直接放行的东西。

两个部分用不同的信号扩缩容。API 按请求速率。Workers 按队列深度，绝不按 CPU——workers 在 15% CPU 下坐着而一百个 job 在等待，基于 CPU 的自动扩缩容会愉快地缩容进积压中。

更好的是：$(queued + running)$ / workers，这在积压形成之前就进行扩缩容。Scale-to-zero 对 workers 是合理的，对 API 则永远不是。

可观测性，三件事。每个进程中每一行日志上的关联 ID $(job_id, tenant)$——当用户说"我的 job 失败了"时，这个过滤器是唯一可行的路径。一套结构化事件词汇：job_submitted、job_started、model_call、tool_call、job_completed、job_failed、job_timeout。

以及带 token、延迟、成本和缓存状态的每步追踪——请求追踪说 job 花了 94 秒；agent 追踪说哪一步占了其中 71 秒。

错误及其修复

按这个顺序构建

两条捷径是合理的：现成的模型网关让你用一跳和一个依赖换到第 5 层，托管 agent 平台让你换到 2、3、4、7 层和第 8 层的一部分，代价是架构是他们的。两者都不能消除你理解你将在其中调试的层的需要，无论哪种方式。

留到以后，大致按顺序：非人工触发器（cron、事件、webhooks——submit 变成内部调用，下游没有任何变化）、带父子记录的扇出 job，以及一个反馈端点，其表会成为你的评估数据集。

这些几乎都与 agent 无关，这与第 1 部分关于 harness 的观点相同。换掉模型、框架或 agent 的工作内容，这个后端不变，因为它的形态源于工作负载：慢、I/O 密集、非确定性、按单位定价。

这四个属性不会消失。

第 3 部分将其部署到 AWS：Fargate、Bedrock、一个不存储凭据的流水线，以及在笔记本电脑上不可能存在的 bug。

上述模式运行在 DevVoice 中，一个 FastAPI、Celery、Redis、Postgres 项目，你可以阅读每个模式背后的代码，包括那些尚未经过验证的部分。

延伸阅读

• Agent Streaming Architecture 2026 · Replay-Then-Tail SSE — 交付层，正确决策
• Prompt Caching Infrastructure · Caching, Fallback, and Load Balancing — 模型层作为一个系统
• Token-Based Rate Limiting for AI Agents · KEDA Autoscaling for Agent Workers — 两个错误的单位：请求计数和 CPU
• The Concurrency Mistake in Every FastAPI AI Service · Temporal's LangGraph Plugin
• API Idempotency for AI Agents · AG-UI Protocol

• 原文链接： x.com/kmeanskaran/status...
• 鸿途知科网 AI 助手，为大家转译优秀英文文章，如有翻译不通的地方，还请包涵～