Today for AIAI 情报与实践
指南返回指南

AI Agent 开发学习与面试指南：01-Python 异步后端与工程基础

讲解生产级 Agent 后端的 asyncio 异步并发机制、Semaphore 限流、调用超时控制、指数退避加随机抖动重试以及写 Tool 时的幂等性设计与面试考点。

本文目录20 个章节

第 1 章：Python 异步后端与工程基础

招聘岗位里的“精通 Python”通常不是指会写数据处理脚本，而是能写可维护、可测试、可观察、能承受并发的服务。Agent 系统同时访问模型 API、向量库、业务数据库和外部工具，绝大多数等待发生在网络 I/O，因此异步、超时、并发控制和错误处理是基础能力。

1. 同步、并发与并行

先区分三个概念：

  • 同步执行：当前调用结束后才执行下一步。
  • 并发：多个任务在同一时间段交替推进，适合大量 I/O 等待。
  • 并行：多个 CPU 核心同时计算，适合 CPU 密集型任务。

模型 API 请求通常属于 I/O 密集型。Embedding 批处理如果调用远程服务也是 I/O；如果在本地运行模型，则可能是 CPU/GPU 密集型。asyncio 能减少等待浪费，但不会让本地矩阵计算自动变快。

FastAPI 官方建议：第三方库提供可等待接口时使用 async def 和 await；阻塞库应放到普通 def 路由或线程池中，避免阻塞事件循环。参见 FastAPI 异步文档。

🔥 P0 高频必会：async 是否一定更快？
不是。异步提升的是 I/O 等待期间的并发利用率。CPU 密集任务仍需要多进程、原生并行库、GPU 或任务队列。把阻塞函数直接放进 async def，反而会卡住整个事件循环。

2. 一个可控并发的模型调用器

下面的例子展示四个生产关键点：并发上限、超时、重试和结构化错误。

PYTHON
import asyncio
import random
from dataclasses import dataclass
from typing import Protocol


class ModelClient(Protocol):
    async def generate(self, prompt: str) -> str: ...


@dataclass
class ModelCallError(Exception):
    code: str
    message: str
    retryable: bool


class ReliableModelGateway:
    def __init__(
        self,
        client: ModelClient,
        max_concurrency: int = 8,
        timeout_seconds: float = 20.0,
        max_attempts: int = 3,
    ) -> None:
        self.client = client
        self.semaphore = asyncio.Semaphore(max_concurrency)
        self.timeout_seconds = timeout_seconds
        self.max_attempts = max_attempts

    async def generate(self, prompt: str) -> str:
        async with self.semaphore:
            for attempt in range(1, self.max_attempts + 1):
                try:
                    async with asyncio.timeout(self.timeout_seconds):
                        return await self.client.generate(prompt)
                except TimeoutError:
                    error = ModelCallError(
                        code="MODEL_TIMEOUT",
                        message=f"model timed out after {self.timeout_seconds}s",
                        retryable=True,
                    )
                except ConnectionError as exc:
                    error = ModelCallError(
                        code="MODEL_NETWORK_ERROR",
                        message=str(exc),
                        retryable=True,
                    )
                except ValueError as exc:
                    raise ModelCallError(
                        code="MODEL_BAD_REQUEST",
                        message=str(exc),
                        retryable=False,
                    ) from exc

                if attempt == self.max_attempts or not error.retryable:
                    raise error

                base = 0.25 * (2 ** (attempt - 1))
                await asyncio.sleep(base + random.uniform(0, 0.1))

        raise RuntimeError("unreachable")

这里有几个容易被忽略的点：

  1. Semaphore 限制的是同时进入外部模型调用的请求数，防止触发限流或耗尽连接池。
  2. 超时必须在调用方设置，不能假设供应商一定会及时返回。
  3. 只重试瞬时错误。参数非法、权限不足、内容策略拒绝通常不应盲目重试。
  4. 重试会放大下游压力，因此要有最大次数、退避和全链路截止时间。

🔥 P0 高频必会：为什么要指数退避加随机抖动？
固定间隔会让同时失败的请求在同一时间再次冲击下游。指数退避降低重试频率，随机抖动打散重试时间。回答时还应提到最大重试次数、仅重试幂等操作和全局超时预算。

3. FastAPI 服务边界

API 层不应该直接塞满 Agent 逻辑。推荐分成四层：

TEXT
HTTP/API 层 -> 应用编排层 -> 领域/工具层 -> 基础设施适配层
  • API 层：认证、参数校验、请求 ID、HTTP 状态码。
  • 应用层：启动 Agent Run、读取状态、取消任务。
  • 领域层：业务规则、工具权限、审批条件。
  • 基础设施层：模型、数据库、向量库、队列的具体客户端。

一个最小接口：

PYTHON
from typing import Literal
from uuid import UUID, uuid4

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field

app = FastAPI()


class RunRequest(BaseModel):
    query: str = Field(min_length=1, max_length=4000)
    mode: Literal["read_only", "allow_write"] = "read_only"


class RunAccepted(BaseModel):
    run_id: UUID
    status: Literal["accepted"] = "accepted"


@app.post("/v1/runs", response_model=RunAccepted, status_code=202)
async def create_run(
    body: RunRequest,
    idempotency_key: str = Header(min_length=8),
) -> RunAccepted:
    if body.mode == "allow_write" and not idempotency_key:
        raise HTTPException(status_code=400, detail="missing idempotency key")

    run_id = uuid4()
#    # 将长任务写入队列，不要让 HTTP 请求无限等待。
    return RunAccepted(run_id=run_id)

为什么返回 202

长时间 Agent Run 可能需要几十秒甚至数分钟。同步等待会占用连接，并让重试语义复杂。常见做法是：

  1. POST /runs 返回 202 Accepted 和 run_id；
  2. 后台 Worker 执行；
  3. 客户端通过轮询、SSE 或 WebSocket 获取进度；
  4. 支持取消、超时与人工审批。

短、稳定、只读的请求可以同步返回；复杂任务更适合异步任务模型。

4. 幂等性：Agent 工具最重要的后端概念之一

如果网络超时，调用方不知道“创建工单”到底成功没有。直接重试可能创建两张工单。解决方式是让相同业务请求多次执行只产生一次效果。

PYTHON
from dataclasses import dataclass


@dataclass
class CreateTicketCommand:
    idempotency_key: str
    user_id: str
    title: str
    description: str


async def create_ticket(command: CreateTicketCommand, repository) -> str:
    existing = await repository.find_by_idempotency_key(command.idempotency_key)
    if existing:
        return existing.ticket_id

#    # 数据库中应给 idempotency_key 加唯一索引，防并发重复写入。
    ticket = await repository.insert(command)
    return ticket.ticket_id

真正可靠的实现还需要数据库唯一约束或事务，仅靠“先查再写”会有并发竞态。

🔥 P0 高频必会：重试和幂等是什么关系？
重试用于恢复瞬时失败，幂等用于保证重试不会重复产生副作用。只读操作天然更接近幂等；写操作要使用幂等键、唯一约束、状态机或补偿事务。

5. 不要阻塞事件循环

错误示例：

PYTHON
import time

async def bad_handler():
    time.sleep(5)  # 阻塞整个事件循环

如果必须调用同步阻塞库，可以临时放入线程：

PYTHON
import asyncio

async def parse_pdf(path: str) -> str:
    return await asyncio.to_thread(blocking_pdf_parser, path)

但线程池不是无限资源。大量文档解析、OCR 或本地推理应放到独立 Worker/队列，并配置并发上限。

6. 错误分类与恢复策略

错误类型示例建议策略
瞬时错误429、连接重置、短暂 5xx有限重试、退避、熔断
参数错误JSON 不合法、上下文过长不重试；修正输入或让模型重构参数
用户可修复缺少订单号、意图不明确暂停并向用户追问
权限错误无权发送邮件或改数据库拒绝、记录审计，不自动扩大权限
业务冲突工单已关闭、库存不足返回结构化业务错误，让 Agent 改计划
未知异常代码缺陷、依赖异常记录 Trace，失败退出并报警

不要把所有异常转成字符串后再次喂给模型。先分类，再决定是系统重试、模型纠正、用户介入还是直接失败。

7. 测试一个 Agent 工具

测试时不要真的调用外部系统。使用依赖注入传入 Fake：

PYTHON
import pytest


class FakeRepository:
    def __init__(self):
        self.items = {}

    async def find_by_idempotency_key(self, key):
        return self.items.get(key)

    async def insert(self, command):
        ticket = type("Ticket", (), {"ticket_id": "T-001"})()
        self.items[command.idempotency_key] = ticket
        return ticket


@pytest.mark.asyncio
async def test_create_ticket_is_idempotent():
    repo = FakeRepository()
    command = CreateTicketCommand("req-12345", "u-1", "登录失败", "无法登录")

    first = await create_ticket(command, repo)
    second = await create_ticket(command, repo)

    assert first == second == "T-001"
    assert len(repo.items) == 1

面试时如果你能展示“相同工具调用不会重复写入”的测试，比只展示成功截图更有说服力。

8. 本章练习

练习 A：并发网关

实现一个批量模型调用函数：

  • 最多并发 5 个请求；
  • 每个请求 10 秒超时；
  • 429 最多重试 3 次；
  • 返回成功结果和失败原因，不因一个失败取消全部任务。

验收：用 Fake Client 模拟延迟、429 和永久失败，写 5 个测试。

练习 B：异步任务 API

实现 /runs、/runs/{id}、/runs/{id}/cancel 三个接口。状态至少包含：queued、running、waiting_approval、succeeded、failed、cancelled。

验收：非法状态迁移必须被拒绝，例如 succeeded -> running。

练习 C：可靠写工具

实现“创建工单”工具：

  • Pydantic 参数校验；
  • 幂等键；
  • 数据库唯一约束的设计说明；
  • 权限检查；
  • 审计日志；
  • 超时后可安全重试。

9. 面试高频问答

🔥 P0：async def 中调用同步阻塞函数会怎样？

它会阻塞事件循环，使同一 Worker 上其他协程无法推进。解决方法是使用异步客户端、短期放入线程池，或将重 CPU/阻塞任务放入独立进程与任务队列。还应配置超时和并发上限。

🔥 P0：如何设计模型 API 的重试？

先分类错误；只重试瞬时且安全的错误；使用指数退避与抖动；设置最大次数和全链路截止时间；写操作依赖幂等性；持续失败时熔断或切换备用模型。

🔥 P0：HTTP 请求为什么不应一直等待 Agent 完成？

长任务容易超过网关超时，客户端重试会导致重复执行，连接资源也会被长时间占用。可以返回 202 + run_id，后台执行，通过 SSE/WebSocket/轮询更新状态。

⭐ P1：并发上限应该放在哪里？

至少有三层：入口限流保护整个服务；模型/工具级 Semaphore 保护单个下游；队列 Worker 并发保护后台资源。仅在入口限流不能防止单次请求内部扇出过大。

⭐ P1：如何防止重复创建业务数据？

使用业务幂等键、数据库唯一约束、事务和明确状态机。外部系统不支持幂等时，可保存请求与外部结果映射，或使用补偿操作，但无法仅靠 Prompt 保证。

10. 本章完成标准

  • 能解释 I/O 并发和 CPU 并行的区别；
  • 能写包含超时、重试、退避和并发控制的异步调用；
  • 能设计异步任务状态机；
  • 能实现并测试幂等写工具；
  • 能说明何时使用线程池、进程、队列或 GPU 服务。

REFERENCES

参考链接

  1. 01FastAPI 异步文档

所属系列

AI Agent 开发学习与面试指南

下一步

继续浏览相关主题

沿着同一主题继续阅读。

查看最新资讯