《AI Agent 工程实战》8.2 背压与资源池：信号量限并发

8.2 背压与资源池：信号量限并发

8.1 证明了并发能省时间，但也留了个尾巴：TaskGroup 会把所有任务一次性全起出来。一轮十个工具调用还好，可当后台有 200 个 run 在排队、每个 run 又要打模型 API 和 Postgres，无限制并发就会同时打开几百个连接、几百个在途请求。下游要么限流把你拒了，要么连接池耗尽直接报错。

本节把 ResearchCopilot 推进到「并发可控」：用 asyncio.Semaphore 给每一类下游设并发上限，用手写最小连接池做连接复用，用有界队列实现背压，真跑 200 个任务验证峰值并发被压在 10，并给出并发上限的估算公式。

8.2.1 无限制并发会打爆下游

并发不是越多越好，因为下游的容量是有限的。三个最常见的容量上限：

• 模型 API 的 RPM / TPM 限制：Claude 按请求数和 token 数限流，超了直接返回 429。
• 数据库连接池上限：Postgres 每个连接占一个后端进程，连接数打满后新连接被拒。
• 第三方服务的 QPS：检索 API、网页抓取服务通常有明确的每秒请求上限。

先量一下「不限并发」到底有多疯。200 个任务、每个睡 0.03s，不加任何限制：

async def unbounded(i, peak):
    peak["cur"] += 1
    peak["max"] = max(peak["max"], peak["cur"])
    await asyncio.sleep(0.03)
    peak["cur"] -= 1

async def main():
    peak = {"cur": 0, "max": 0}
    async with asyncio.TaskGroup() as tg:
        for i in range(200):
            tg.create_task(unbounded(i, peak))
    print(f"不限并发：200 个任务峰值并发 = {peak['max']}")

真跑输出：

不限并发：200 个任务峰值并发 = 200

峰值并发 200。如果这 200 个任务每个都要开一条数据库连接，那就是 200 个连接同时建立——而 Postgres 默认 max_connections 才 100，直接打满并开始拒绝。这不是理论风险，是压测里最常见的第一个坑。

8.2.2 Semaphore：并发闸门

asyncio.Semaphore 是一个计数器：初始化时给一个上限 N，async with sem 进入时减一，退出时加一。计数为零时，后来的协程会在 await sem.acquire() 上排队等待，直到有人释放。它就是并发的闸门——无论外面起多少任务，同时进门的永远不超过 N 个。

import asyncio

async def limited_fetch(i, sem, state):
    async with sem:                       # 并发闸门：超过上限就在这里排队
        state["cur"] += 1
        state["peak"] = max(state["peak"], state["cur"])
        await asyncio.sleep(0.05)         # 模拟下游调用耗时
        state["cur"] -= 1
        return i

async def main():
    N, LIMIT = 200, 10
    sem = asyncio.Semaphore(LIMIT)        # 最多 10 个并发
    state = {"cur": 0, "peak": 0}
    t0 = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        for i in range(N):
            tg.create_task(limited_fetch(i, sem, state))
    dt = time.perf_counter() - t0
    print(f"任务数={N} 并发上限={LIMIT} 峰值并发={state['peak']} 总耗时={dt:.2f}s")
    print(f"理论下界={N/LIMIT*0.05:.2f}s")

注意 async with sem 的位置：它包住了「进入 → 干活 → 退出」的整个临界区。如果把 async with sem 只包住「计数」那一行，干活在闸门外，闸门就形同虚设。闸门要包住真正占用下游资源的那段。

8.2.3 真跑：200 个任务，上限 10

真跑上面这段（Python 3.14.6，time.perf_counter 计时）：

任务数=200 并发上限=10 峰值并发=10 总耗时=1.03s
理论下界=1.00s

逐项读：

• 峰值并发 = 10：闸门确实起作用了，从 8.2.1 的 200 被压到了 10，下游最多同时看到 10 个请求。
• 总耗时 1.03s：200 个任务、每个 0.05s、并发 10，理论耗时是 200 / 10 × 0.05 = 1.00s，实测 1.03s，多出的 0.03s 是任务调度开销。说明任务被均匀地分批执行了，没有空转。
• 对比 8.2.1：不加限流峰值 200、耗时约 0.03s——快了，但代价是打爆下游。限流是用吞吐换稳定。

这就是「并发上限」的价值：它不是让系统更快，而是让系统在可控的压力下稳定运行。省下来的那 0.9s 换不来任何东西，但省掉的那 190 个并发连接能救一次线上事故。

8.2.4 资源池：复用比限制更重要

信号量解决「同时最多几个」，但每次调用都新建连接再关掉仍然是浪费——建连握手（TCP + TLS + 认证）可能比查询本身还慢。资源池解决的是「复用」：预先建好一批连接，借出去、还回来，反复使用。

psycopg_pool 是生产首选，但它的接口本质很简单，手写一个最小版本就能看清原理：

from collections import deque

class MiniPool:
    """手写最小异步连接池：固定 size，借出/归还，复用连接。"""
    def __init__(self, size: int, connect):
        self._connect = connect
        self._free = deque()              # 空闲连接栈
        self._created = 0
        self._lock = asyncio.Lock()
        self._sem = asyncio.Semaphore(size)

    async def acquire(self):
        await self._sem.acquire()         # 池满时在此排队（这就是上限）
        async with self._lock:            # 保护 _free 与 _created 的并发访问
            if self._free:
                return self._free.pop()   # 复用已有连接
            self._created += 1
            return await self._connect(self._created)   # 池未满才新建

    async def release(self, conn):
        async with self._lock:
            self._free.append(conn)       # 归还
        self._sem.release()

真跑：12 个任务、池大小 3：

12 个任务 / 池大小 3：实际建连 3 次，峰值并发 3，耗时 0.19s

关键数字是实际建连 3 次——12 个任务只建了 3 条连接，其余 9 次全是复用。如果不用池，就是 12 次建连。池的 size 同时兼任两个角色：并发上限（Semaphore）和连接数上限（_free 最多存 size 条）。这正是它比裸信号量更好的地方——限并发和管资源用同一套机制。

两个细节值得点出：asyncio.Lock 保护的是 _free 和 _created 这两个共享可变状态（对应 8.1 那张表里「不能并发」的项）；release 里先 append 再 release() 信号量，顺序不能反，否则会出现「信号量放行了但连接还没还回去」的窗口。

生产里直接用 psycopg_pool.AsyncConnectionPool（本机实测 psycopg_pool 3.3.3），接口和上面几乎一样：

from psycopg_pool import AsyncConnectionPool

pool = AsyncConnectionPool(
    conninfo="postgresql://postgres:postgres@localhost:15435/research",
    min_size=2, max_size=10, open=False,
)
await pool.open()
async with pool.connection() as conn:      # 借出，退出时自动归还
    cur = await conn.execute("SELECT 1")
    print(await cur.fetchone())

对着本机的 Postgres 17.11 容器真连真跑：

psycopg_pool 3.3.3
SELECT 1 -> (1,)
pool size: 3 max: 10

pool_size: 3 说明池按需建了 3 条连接（min_size=2 起手，用到第三条），max: 10 是上限——同一个 max_size 同时管着并发和连接数，和手写版一个道理。

连接串照本卷口径是 localhost:15435（书稿专用容器，端口非默认，本机 5432 已被占用）。读者要自己起容器才能连：

docker run --name ab-pg -e POSTGRES_PASSWORD=postgres \
  -e POSTGRES_DB=research -p 15435:5432 -d postgres:17-alpine

没起容器时，上面的 MiniPool 演示（用假连接）不依赖数据库，可以直接跑。

8.2.5 背压：让上游慢下来，而不是无限堆积

信号量和连接池解决「下游别被打爆」，背压解决的是上游生产快于下游消费的问题。想象一个 worker：它从队列取 run，每个 run 要处理几千个 token 事件往外推。如果生产者（模型）吐 token 的速度快过消费者（写库 / 推 SSE）的速度，中间就会无限堆积——内存涨、延迟涨，最后 OOM。

背压的核心思想是：当消费跟不上时，让生产端慢下来，而不是把积压堆在内存里。在 asyncio 里，用有界队列就能实现——asyncio.Queue(maxsize=N) 的 put 在队列满时会阻塞生产者：

async def main():
    q = asyncio.Queue(maxsize=5)      # 有界队列：上限 5
    peak_q = 0
    async def producer():
        nonlocal peak_q
        for i in range(50):
            await q.put(i)            # 队列满则在此阻塞 —— 这就是背压
            peak_q = max(peak_q, q.qsize())
        await q.put(None)             # 哨兵：通知消费者收工
    async def consumer():
        while True:
            item = await q.get()
            if item is None:
                break
            await asyncio.sleep(0.01) # 消费比生产慢
    await asyncio.gather(producer(), consumer())
    print(f"有界队列 maxsize=5 峰值深度={peak_q}")

真跑（队列 maxsize=5）：

有界队列 maxsize=5 峰值深度=5

峰值深度恰好等于 maxsize，从没超过 5。如果换成无界队列（maxsize=0），50 个元素会瞬间全部堆进去，峰值深度就是 50——队列本身成了内存泄漏。有界队列把「积压」变成了「阻塞」：生产者在满队列前停下，等消费者腾出空间，整个系统的在途量被钉死在 maxsize 这个常数上。

这个「让上游慢下来」的信号会一路向上传播：队列满 → 模型输出协程阻塞 → 该 run 的处理变慢 → 队列消费变慢 → 新 run 在更上游排队。最终用户看到的可能是「响应变慢」，但系统不会崩。这比「响应很快然后突然 OOM」好得多——可控的慢，胜过不可控的死。

8.2.6 并发上限怎么估

Semaphore(10) 里的 10 不是拍脑袋来的，它可以按下游容量反推。核心公式是利特尔法则：

并发上限 ≈ 下游允许的 QPS × 单次调用的平均耗时（秒）

推导很直观：如果下游每秒最多吃 QPS 个请求，而每个请求平均占用一个并发槽 L 秒，那么稳态下需要同时在途的槽数就是 QPS × L。超过这个数，多出来的并发只会排队或被拒。

按这个公式估几个典型下游：

几条经验：

• 留余量：算出 80 就设 40–60，别贴着上限跑——下游是共享的，你的估算有误差，别人的流量也在抢。
• 分类设上限：模型 API、数据库、检索各有各的 Semaphore，不要用一个全局闸门——它们的瓶颈完全不同。
• 上限随负载调：上限是配置项（第 2 章的分层配置），压测出拐点后（第 14 章）再调，而不是写死在代码里。
• 超时兜底：限流和超时是配套的——被限流的请求不能无限等，必须有超时（下一节）。

8.2.7 小结与验收

本节给 8.1 的并发装上了三道刹车：

• 信号量限并发：200 个任务峰值并发从 200 压到 10，用吞吐换稳定。
• 资源池复用：12 个任务只用 3 条连接，建连开销被摊薄。
• 背压：有界队列把积压变成阻塞，在途量被钉死在常数。

验收清单：

• 每一类下游（模型 / 数据库 / 检索）有独立的并发上限
• 并发上限不是拍脑袋，是按 QPS × 耗时 估出来的，且留了余量
• 连接池的 max_size 与数据库的 max_connections 匹配（别设成 200 打爆 Postgres）
• 生产者 → 消费者的中间队列是有界的，没有无界队列
• 被限流的请求有超时兜底，不会无限排队

真实环境 / 真实模型下会怎样：本节的「下游」都是 asyncio.sleep 假装的，容量是确定的。真实下游的容量会动态变化——模型 API 在高峰期更早限流，数据库在别的服务也在用时连接更紧张。所以并发上限应该是可运行时调整的配置，而不是编译期常量。另一个真实差异是限流信号：假下游只会让你等，真下游会返回 429 / 连接拒绝，这时光靠信号量不够，还要有退避重试（第 7.2 节）——被拒的请求不能立刻重发，否则限流会更严重。最后，背压在真实系统里往往跨进程：worker 和 API 进程之间隔着 Redis 队列，真正的背压要靠队列长度监控 + 拒绝新任务来实现，而不仅仅是内存里的 asyncio.Queue。

阅读导航：上一节：8.1 引擎内并发：并行工具与并行子 Agent · 下一节：8.3 取消、超时与泄漏：慢工具注入验证止损 。