Pydantic AI 流式请求卡死的修复方法

核心结论

Pydantic AI 2.10.0 至 2.52.x 的模型级流式并发限制存在槽位泄漏：使用 ConcurrencyLimitedModel 或 limit_model_concurrency 时，流已经结束或被取消，并发槽位却可能没有归还，最终让共享同一限制器的后续请求一直等待。直接修复方式是把 pydantic-ai 或 pydantic-ai-slim 升级到 2.53.0 及以上；无法立即升级时，应暂时改用 Agent 级 max_concurrency，或者停止让模型级限制器承载流式请求。

这个问题容易被误判为上游模型变慢、SSE 断流、网络抖动或 API 渠道限流。区别在于：上游请求可能已经结束，本地进程中的并发计数却只增不减。排查外部接口前，可以先参考站内的 LLM 流式输出工程排查，再检查 Pydantic AI 的版本、限制器所在层级和 running_count。如果还需要核对不同 API 服务的接入方式，可从模型与平台目录查看对应文档入口，但本次故障的根因位于客户端并发控制，不是模型能力或供应商状态。

哪些配置会受影响

官方安全公告给出的受影响范围是 pydantic-ai 与 pydantic-ai-slim 的 >=2.10.0,<2.53.0 版本。问题需要同时满足两个条件：一是把并发限制放在模型包装层，二是使用流式请求。典型写法如下：

from pydantic_ai import Agent, ConcurrencyLimiter
from pydantic_ai.models.concurrency import ConcurrencyLimitedModel

limiter = ConcurrencyLimiter(max_running=5)
model = ConcurrencyLimitedModel("provider:model", limiter=limiter)
agent = Agent(model)

受影响的不只是客户端主动断开连接。官方公告列出的触发路径包括：消费者提前停止迭代、消费代码抛出异常、任务被取消，以及使用默认 debounce_by=0.1 完整消费 stream_text()。因此，“每次都把流读完”不能作为旧版本的安全保证。修复 PR 的回归测试还覆盖了显式关闭生成器、进入流后不消费、run_stream_events() 提前退出和节点流提前退出等路径。

两个范围需要明确排除。第一，Agent 级 max_concurrency 不受这个缺陷影响。第二，非流式模型请求不受影响。如果服务只调用 agent.run()，或者并发限制只配置在 Agent(..., max_concurrency=...)，就不属于公告描述的漏洞路径。不过，仍应检查是否同时在 Agent 层和模型层复用了同一个限制器，因为 2.53.0 对这种配置增加了显式拒绝。

为什么流结束了，槽位却没有归还

旧实现使用 AnyIO 的 CapacityLimiter。AnyIO 的接口把令牌借给当前任务：acquire() 为当前任务取得令牌，release() 则释放当前任务持有的令牌；如果执行释放的任务并未借到令牌，接口会抛出 RuntimeError。这种任务所有权适合“同一个异步任务获取并释放”的生命周期。

Pydantic AI 的流式生命周期并不总在同一个任务内完成。打开模型流、执行去抖预取、消费输出和关闭流可能由不同内部任务处理。旧版本可能在任务甲取得槽位，却在任务乙清理流。此时 AnyIO 拒绝释放，错误信息可能是：

RuntimeError: this borrower isn't holding any of this CapacityLimiter's tokens

更隐蔽的后果不是这条异常，而是槽位仍被记为占用。假设 max_running=5，连续五次触发泄漏后，第六个共享限制器的请求就可能永远排队。网络客户端反复开始流并断开，也能逐步耗尽一个长期存活的共享限制器，所以官方将其归类为可用性问题。

2.53.0 把内置实现改成了 AnyIO Semaphore。信号量只记录可用槽位数量，不把每个槽位绑定到取得它的具体任务，因而允许流在另一任务中关闭并归还槽位。修复后的公开契约也明确要求：自定义 AbstractConcurrencyLimiter.release() 必须允许在不同于 acquire() 的任务中执行。

用最小测试确认是否发生泄漏

先确认实际安装版本，不要只看锁文件或镜像构建参数：

python -c "from importlib.metadata import version; print(version('pydantic-ai'))"
python -c "from importlib.metadata import version; print(version('pydantic-ai-slim'))"

如果应用只安装了其中一个包，另一条命令报“未找到分发包”可以忽略。随后搜索配置中是否出现模型级限制器：

rg "ConcurrencyLimitedModel|limit_model_concurrency" .
rg "max_concurrency" .

下面的测试改写自 2.53.0 的官方回归用例，不需要调用外部模型。它使用 FunctionModel 生成三个流式片段，并在完整消费后检查限制器计数：

import asyncio

from pydantic_ai import Agent, ConcurrencyLimiter
from pydantic_ai.models.concurrency import ConcurrencyLimitedModel
from pydantic_ai.models.function import FunctionModel


async def stream_function(messages, info):
    for chunk in ("a", "b", "c"):
        yield chunk


async def main():
    limiter = ConcurrencyLimiter(max_running=1)
    model = ConcurrencyLimitedModel(
        FunctionModel(stream_function=stream_function),
        limiter=limiter,
    )
    agent = Agent(model)

    async with agent.run_stream("hi") as result:
        async for _ in result.stream_text():
            pass

    print("running_count=", limiter.running_count)
    print("waiting_count=", limiter.waiting_count)


asyncio.run(main())

修复后的预期结果是 running_count=0 和 waiting_count=0，随后再次运行流式请求仍能完成。旧版本可能在流结束处抛出借用者相关的 RuntimeError，也可能留下非零运行计数。生产环境还可以给这两个计数加监控：请求已经结束，但 running_count 长时间维持在上限且 waiting_count 持续增加，是本地槽位未释放的重要信号。

不要把“增加请求超时”当成修复。取消任务正是受影响路径之一，旧版本中更短的超时可能让流更频繁地进入跨任务清理。也不要只重启进程；重启会清空内存中的限制器状态，却不会消除下一轮泄漏。

升级与配置迁移

安装修复版时，应更新实际部署使用的包，并重新生成锁文件：

python -m pip install "pydantic-ai>=2.53.0"
# 使用 slim 发行包的项目改为：
python -m pip install "pydantic-ai-slim>=2.53.0"

如果暂时不能升级，官方建议有两条临时路径。其一，把并发限制移到 Agent 层：

from pydantic_ai import Agent

agent = Agent("provider:model", max_concurrency=5)

其二，保留模型级限制器，但不要通过它执行流式请求。临时方案需要结合应用行为选择；如果业务必须持续输出流，Agent 级限制比“禁用流式”更接近原来的用户体验。

升级后还要处理兼容性变化。不要把同一个限制器同时交给 Agent 和它内部的 ConcurrencyLimitedModel：

from pydantic_ai import Agent, ConcurrencyLimiter
from pydantic_ai.models.concurrency import ConcurrencyLimitedModel

shared = ConcurrencyLimiter(max_running=5)
model = ConcurrencyLimitedModel("provider:model", limiter=shared)

# 错误：同一个限制器同时用于 Agent 层和模型层。
agent = Agent(model, max_concurrency=shared)

2.53.0 会对这种共享方式抛出 UserError，因为同一请求可能重复取得同一个池的槽位并造成死锁。嵌套的多个 ConcurrencyLimitedModel 也不能共享同一个限制器。应当只在一个层级限流，或者确有两个独立资源边界时使用不同限制器。

自定义限制器同样要审查。新契约要求每次成功的 acquire() 都对应一次 release()，同一任务再次调用 acquire() 也会再占用一个槽位，而且释放可能来自另一个任务。若自定义实现依赖任务本地变量、当前任务 ID 或 AnyIO CapacityLimiter.release() 的任务所有权，就需要改成允许跨任务归还的设计，并补充“完整消费、提前退出、异常、取消”四类测试。

上线前的验证顺序

建议按下面的顺序完成修复，而不是只确认安装命令执行成功：

在运行容器或虚拟环境内读取包版本，确认不是旧镜像或旧依赖层。
执行最小流式测试，确认一次完整消费后 running_count 回到零。
分别测试提前 break、消费函数抛异常和取消任务，确认槽位都能归还。
搜索 Agent 层、模型包装层和嵌套包装层，确保同一个限制器没有被重复使用。
如果实现了 AbstractConcurrencyLimiter 子类，验证跨任务 release()，不要只跑同任务的单元测试。
在预发布环境连续运行超过 max_running 次流式请求，确认后续请求不会永久等待。
观察 running_count、waiting_count、请求完成数和取消数，确保计数随请求结束下降。

一句话判断是否修好：流由哪个任务结束并不重要，所有结束路径都必须归还槽位。Pydantic AI 2.53.0 修复了内置限制器，但应用仍要消除重复分层和不兼容的自定义实现，才能避免把本地死锁继续误报成上游 API 故障。

参考来源

以上来源均于 2026 年 10 月 4 日打开并核对原文。