Asyncio调用Claude时偶现‘RuntimeError: Event loop is closed’错误求助

问题 #

编写了一个使用Claude和asyncio的测试示例，循环多次调用asyncio.run()，迭代中偶尔抛出RuntimeError: Event loop is closed异常，代码表面运行正常但无法定位问题。示例代码如下：

import asyncio
from time import perf_counter

import numpy as np
from anthropic import Anthropic, AsyncAnthropic


async def coroutine1():
    prompt = "Who discovered gravity. Answer in 10 words."

    client = AsyncAnthropic()
    message = await client.messages.create(
        model="claude-3-5-sonnet-20240620",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    )

    return message.content[0].text


async def coroutine2():
    prompt = "Who discovered radioactivity. Answer in 10 words."

    client = AsyncAnthropic()
    message = await client.messages.create(
        model="claude-3-5-sonnet-20240620",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    )

    return message.content[0].text


async def main():
    async with asyncio.TaskGroup() as tg:
        t0 = tg.create_task(coroutine1())
        t1 = tg.create_task(coroutine2())
    print(t0.result())
    print(t1.result())


if __name__ == "__main__":
    for i in range(10):
        tic = perf_counter()
        asyncio.run(main())
        toc = perf_counter()
        print(f"Elapsed time = {toc - tic:.3f}")

完整错误回溯：

Task exception was never retrieved
future: <Task finished name='Task-40' coro=<AsyncClient.aclose() done, defined at /Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpx/_client.py:2024> exception=RuntimeError('Event loop is closed')>
Traceback (most recent call last):
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpx/_client.py", line 2031, in aclose
    await self._transport.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpx/_transports/default.py", line 389, in aclose
    await self._pool.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpcore/_async/connection_pool.py", line 313, in aclose
    await self._close_connections(closing_connections)
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpcore/_async/connection_pool.py", line 305, in _close_connections
    await connection.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpcore/_async/connection.py", line 171, in aclose
    await self._connection.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpcore/_async/http11.py", line 265, in aclose
    await self._network_stream.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpcore/_backends/anyio.py", line 55, in aclose
    await self._stream.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/anyio/streams/tls.py", line 201, in aclose
    await self.transport_stream.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/anyio/_backends/_asyncio.py", line 1287, in aclose
    self._transport.close()
  File "/Users/foo/.pyenv/versions/3.12.1/lib/python3.12/asyncio/selector_events.py", line 1206, in close
    super().close()
  File "/Users/foo/.pyenv/versions/3.12.1/lib/python3.12/asyncio/selector_events.py", line 871, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "/Users/foo/.pyenv/versions/3.12.1/lib/python3.12/asyncio/base_events.py", line 792, in call_soon
    self._check_closed()
  File "/Users/foo/.pyenv/versions/3.12.1/lib/python3.12/asyncio/base_events.py", line 539, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
Task exception was never retrieved
future: <Task finished name='Task-41' coro=<AsyncClient.aclose() done, defined at /Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpx/_client.py:2024> exception=RuntimeError('Event loop is closed')>
Traceback (most recent call last):
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpx/_client.py", line 2031, in aclose
    await self._transport.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpx/_transports/default.py", line 389, in aclose
    await self._pool.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpcore/_async/connection_pool.py", line 313, in aclose
    await self._close_connections(closing_connections)
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpcore/_async/connection_pool.py", line 305, in _close_connections
    await connection.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpcore/_async/connection.py", line 171, in aclose
    await self._connection.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpcore/_async/http11.py", line 265, in aclose
    await self._network_stream.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/httpcore/_backends/anyio.py", line 55, in aclose
    await self._stream.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/anyio/streams/tls.py", line 201, in aclose
    await self.transport_stream.aclose()
  File "/Users/foo/Desktop/async/venv/lib/python3.12/site-packages/anyio/_backends/_asyncio.py", line 1287, in aclose
    self._transport.close()
  File "/Users/foo/.pyenv/versions/3.12.1/lib/python3.12/asyncio/selector_events.py", line 1206, in close
    super().close()
  File "/Users/foo/.pyenv/versions/3.12.1/lib/python3.12/asyncio/selector_events.py", line 871, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "/Users/foo/.pyenv/versions/3.12.1/lib/python3.12/asyncio/base_events.py", line 792, in call_soon
    self._check_closed()
  File "/Users/foo/.pyenv/versions/3.12.1/lib/python3.12/asyncio/base_events.py", line 539, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

原因分析 #

• 每次调用asyncio.run()都会新建事件循环，执行完毕后直接关闭循环。
• AsyncAnthropic客户端基于httpx实现，内部维护连接池。任务完成后客户端的aclose()方法会异步执行，但此时事件循环已经关闭，导致资源清理操作失败报错。
• 原代码中每个协程都创建独立的AsyncAnthropic客户端，没有等待客户端关闭完成就退出了事件循环。

修复方案 #

修改后的代码 #

import asyncio
from time import perf_counter

from anthropic import AsyncAnthropic


async def coroutine1(client):
    prompt = "Who discovered gravity. Answer in 10 words."
    message = await client.messages.create(
        model="claude-3-5-sonnet-20240620",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    )
    return message.content[0].text


async def coroutine2(client):
    prompt = "Who discovered radioactivity. Answer in 10 words."
    message = await client.messages.create(
        model="claude-3-5-sonnet-20240620",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    )
    return message.content[0].text


async def run_iteration(client):
    async with asyncio.TaskGroup() as tg:
        t0 = tg.create_task(coroutine1(client))
        t1 = tg.create_task(coroutine2(client))
    print(t0.result())
    print(t1.result())


async def main():
    # 复用同一个客户端实例，通过async with管理生命周期
    async with AsyncAnthropic() as client:
        for i in range(10):
            tic = perf_counter()
            await run_iteration(client)
            toc = perf_counter()
            print(f"Elapsed time = {toc - tic:.3f}")


if __name__ == "__main__":
    asyncio.run(main())

关键修改点 #

1. 复用客户端实例：将AsyncAnthropic客户端提升到外层，通过async with管理，确保整个循环过程复用同一个客户端，减少连接开销的同时保证关闭逻辑正常执行。
2. 协程接收客户端参数：不再在协程内部创建客户端，避免重复初始化和未完成的资源清理操作。
3. 调整循环逻辑：将10次迭代移到单个事件循环内执行，避免多次创建/关闭事件循环导致的资源冲突。

内容的提问来源于stack exchange，提问作者Vincent Garcia