Phoenix Playground 异步 LLM 客户端生命周期管理统一异步上下文管理器模式的设计与实现【免费下载链接】phoenixAI Observability Evaluation项目地址: https://gitcode.com/gh_mirrors/phoenix13/phoenix本文基于 Phoenix 内部设计规格 async-llm-client-lifecycle.md深入剖析 Phoenix Playground 基础设施中如何通过统一异步上下文管理器工厂模式管理 OpenAI、Azure OpenAI、Anthropic、Google GenAI、AWS Bedrock 等 LLM Provider 客户端的完整生命周期。你将理解 boto3 为何会冻结 asyncio 事件循环、aioboto3 为何强制使用async with、各 SDK 上下文管理器实现的差异以及 Phoenix 在源码中如何落地这一模式——读完即可在自己的异步 LLM 服务中正确设计客户端创建、复用与资源清理策略。概述为什么需要统一的客户端生命周期管理Phoenix 的 Playground模型实验场功能支持在同一个异步服务内调用多个 LLM ProviderOpenAI、Azure OpenAI、Anthropic、Google GenAI、AWS Bedrock每个 Provider 的 SDK 拥有截然不同的客户端生命周期模式。规格文档 async-llm-client-lifecycle.md 描述了在 Phoenix 全部 LLM Provider 客户端上引入统一异步上下文管理器模式unified async context manager pattern的动机与设计保证 HTTP 连接生命周期的正确管理创建、使用、显式关闭提供与底层 SDK 无关的一致接口修复 AWS Bedrock 的 boto3 阻塞事件循环问题。对应实现位于 playground_clients.py客户端工厂的类型定义与各 Provider 的工厂实现位于 model_provider.py。变更内容各 Provider 客户端模式的演进规格文档给出了本次设计的变更对照表Provider变更前变更后影响AWS Bedrockboto3阻塞 I/Oaioboto3异步 I/O修复事件循环阻塞OpenAI每请求新建客户端每请求新建客户端无变化Azure OpenAI每请求新建客户端每请求新建客户端无变化Anthropic每请求新建客户端每请求新建客户端无变化Google GenAI每请求新建客户端每请求新建客户端无变化核心洞察所有 Provider 原本就采用每请求新建客户端fresh client per request的模式因此已经承担了每次请求约 20–250ms 的连接开销。本次改造并未给非 Bedrock Provider 引入新的开销——它只修复了 Bedrock 的阻塞行为。统一工厂模式unified factory pattern以显式异步上下文管理器正式化了既有的 fresh-client 模式确保所有 Provider 的资源清理行为一致。问题陈述各 SDK 客户端生命周期的差异Phoenix 的 Playground 支持多个 LLM Provider每个 Provider 的 SDK 客户端生命周期模式各不相同OpenAI / Azure OpenAI / Anthropic客户端可以同步创建但持有 HTTP 连接应显式关闭AWS Bedrockboto3同步客户端使用阻塞 I/O会冻结异步事件循环Google GenAI同步/异步双客户端生命周期管理相互独立。值得注意的是Azure OpenAI 复用了同一个AsyncOpenAI客户端仅通过自定义base_url指向 Azure 端点形如https://{endpoint}/openai/v1/从而避免了单独的AsyncAzureOpenAI类。在 playground_clients.py 中可以看到 base_url 的构造逻辑# Construct the v1 API base URL endpoint endpoint.rstrip(/) base_url (endpoint if endpoint.endswith(/openai/v1) else f{endpoint}/openai/v1) /为什么 boto3 会阻塞事件循环使用yield配合 boto3 流式输出并不会让它变成异步。yield关键字只是创建生成器以便增量返回数据并不改变底层 I/O 行为# boto3 (BLOCKING) response boto3_client.converse_stream(...) # ← Blocks until server responds for event in response[stream]: # ← Each iteration blocks waiting for next chunk yield event # ← yield returns control, but NEXT iteration blocks again每一次网络读取都是阻塞式 socket 操作。在这些等待期间asyncio 事件循环被冻结其他协程无法执行boto3: [BLOCK 50ms][yield][BLOCK 80ms][yield][BLOCK 60ms]... ↑ ↑ ↑ Event loop Event loop Event loop frozen frozen frozen aioboto3: [await 50ms] [await 80ms] [await 60ms]... ↓ ↓ ↓ Other requests Other requests Other requests can execute can execute can execute实际影响评估低影响场景阻塞可以接受本地开发、单用户使用生产环境并发 Bedrock 流式请求少于 5 个专用 Bedrock-only 部署且没有其他异步工作负载。高影响场景强烈建议使用 aioboto310 个并发 Bedrock 流混合异步工作负载数据库查询、其他 API 调用共享同一个事件循环对尾延迟tail latency敏感的延迟敏感型应用。场景boto3 影响单流约 100 tokens总计约 3-5s 阻塞用户本来就在等待5 个并发流每个流延迟增加约 20-50%10 个并发流显著串行化延迟叠加放大混合工作负载其他异步操作DB、HTTP在 Bedrock I/O 期间被饿死缺少生命周期管理的后果负载下连接池耗尽connection pool exhaustion长运行应用中的资源泄漏resource leaks因连接池耗尽引发的APITimeoutError尤其在流式场景下。为什么 aioboto3 要求异步上下文管理器与 OpenAI/Anthropic SDK可以同步创建客户端、稍后可选关闭不同aioboto3 有更严格的约束这直接驱动了本次设计异步客户端创建aioboto3 的session.client()返回的是ClientCreatorContext而非客户端本身。只有进入异步上下文管理器async with之后才能拿到真正的客户端。这是因为凭证解析、端点发现和 HTTP 会话建立都是异步操作。aiohttp 需要显式清理aiohttp 的__del__对未关闭的会话只会发出ResourceWarning并不会真正关闭连接。不调用__aexit__会遗留已分配的 TCP 连接。由于 aiohttp connector 默认连接数上限为 100未关闭的客户端最终会导致新请求等待可用连接进而引发超时。会话与客户端的生命周期绑定在 boto3 中可以无限期持有客户端而在 aioboto3 中客户端的 HTTP 会话与上下文管理器作用域绑定——退出上下文会将_sessions置为None之后的 API 调用都会失败。这些约束决定了不能简单包装aioboto3 来对齐 OpenAI/Anthropic 模式而是为所有 Provider 采用 aioboto3 的上下文管理器模式从而在所有地方保证一致性与正确的资源清理。设计决策统一工厂模式Unified Factory Pattern所有 Provider 使用一个返回异步上下文管理器的工厂可调用对象factory callable。工厂在闭包中捕获必要的配置每次请求创建全新客户端。规格文档给出了核心骨架ClientT TypeVar(ClientT) class PlaygroundClient(ABC, Generic[ClientT]): _client_factory: Callable[[], AsyncContextManager[ClientT]] async def chat_completion_create(self, ...): async with self._client_factory() as client: # client is typed as ClientT # Provider-specific logic using client ... # Subclasses specify their client type class OpenAICompatibleClient(PlaygroundClient[AsyncOpenAI]): ... class AnthropicClient(PlaygroundClient[AsyncAnthropic]): ... class GoogleClient(PlaygroundClient[GoogleAsyncClient]): ... class BedrockClient(PlaygroundClient[BedrockRuntimeClient]): ...这一骨架在仓库中的真实实现位于 playground_clients.pyPlaygroundClient(ABC, Generic[ClientT])基类接收client_factory: ClientFactory[ClientT]、model_name、provider三个参数其中_client_factory类型为ClientFactory[ClientT]。所有子类通过register_llm_client装饰器注册到 Playground 注册表playground_registry.py。工厂类型ClientFactory 与 LLMClientFactory在 model_provider.py 中定义了工厂协议及其实现class ClientFactory(Protocol[ClientT_co]): def __call__(self) - AbstractAsyncContextManager[ClientT_co]: ... property def rate_limit_key(self) - Hashable: ... class LLMClientFactory(ClientFactory[ClientT]): Factory for creating LLM clients with rate limit key for bucketing. __slots__ (_create, _rate_limit_key) def __init__(self, create: Callable[[], AbstractAsyncContextManager[ClientT]], rate_limit_key: Hashable) - None: self._create create self._rate_limit_key rate_limit_key def __call__(self) - AbstractAsyncContextManager[ClientT]: return self._create() property def rate_limit_key(self) - Hashable: return self._rate_limit_keyLLMClientFactory将创建客户端的可调用对象与用于限流分桶的rate_limit_key配对。各 Provider 的限流键构造函数也集中在此文件中例如openai_rate_limit_key(api_key, base_url)、azure_rate_limit_key(endpoint, credential)、anthropic_rate_limit_key(api_key, base_url)、bedrock_rate_limit_key(region, credential)、google_rate_limit_key(api_key, base_url)。PlaygroundClient.get_rate_limit_key()直接委托给工厂BedrockClient与OpenAIChatCompletionsClient等子类还会叠加model_name以体现按模型分桶的限流语义见 playground_clients.py。各 Provider 的工厂实现Provider工厂实现OpenAIlambda: AsyncOpenAI(api_keyapi_key, ...)Azure OpenAIlambda: AsyncOpenAI(api_keyapi_key, base_urlazure_base_url, ...)Anthropiclambda: AsyncAnthropic(api_keyapi_key, ...)Google GenAIlambda: Client(api_keyapi_key).aioAWS Bedrocklambda: session.client(bedrock-runtime, ...)在 playground_clients.py 的_get_builtin_provider_client中可以看到各内置 Provider 的真实工厂构造OpenAIL3363-L3373AsyncOpenAI(api_keyapi_key, base_urlbase_url, default_headersheaders, timeout30)并包装为LLMClientFactory(create_openai_client, openai_rate_limit_key(...))Azure OpenAIL3413-L3420AsyncOpenAI(api_keyapi_key, base_urlbase_url, default_headersheaders)若没有 API key则使用azure.identity.aio的DefaultAzureCredentialget_bearer_token_provider生成 token provider以api_keytoken_provider传入要求openai1.106.0见 L3429-L3444——这比已废弃的AsyncAzureOpenAI类更简洁AnthropicL3475-L3480anthropic.AsyncAnthropic(api_keyapi_key, default_headersheaders)其本身实现了__aenter__/__aexit__天然是异步上下文管理器Google GenAIL3516-L3524由于 Google 的AsyncClient.__aexit__签名不符合AbstractAsyncContextManager协议返回None而非bool | None源码用asynccontextmanager显式包装asynccontextmanager async def create_google_client() - AsyncIterator[GoogleAsyncClient]: async with GoogleGenAIClient(api_keyapi_key).aio as client: yield clientAWS BedrockL3564-L3583先构造aioboto3.Session(...)显式传入 access key / secret key / session token 与 region工厂返回aioboto3_session.client(service_namebedrock-runtime)——即ClientCreatorContext必须通过async with进入def create_bedrock_client() - AbstractAsyncContextManager[BedrockRuntimeClient]: return aioboto3_session.client(service_namebedrock-runtime) bedrock_client_factory: ClientFactory[BedrockRuntimeClient] LLMClientFactory( create_bedrock_client, bedrock_rate_limit_key(region, aws_access_key_id) )对于内置 Provider凭证解析遵循三级优先级请求显式凭证 → 数据库加密 Secret → 进程环境变量见_resolve_provider_api_keyL3185-L3217。规格文档还特别指出Azure OpenAI 复用AsyncOpenAI加自定义base_url该方式同样适用于把 Azure AD token provider 作为api_key传入的场景。消费模式Consumption Pattern所有 Provider 使用一致的消费方式async def chat_completion_create(self, messages, tools, **params): async with self._client_factory() as client: # For OpenAI/Azure/Anthropic: Wrap httpx client for instrumentation client._client _HttpxClient(client._client, self._attributes) # Provider-specific API calls response await client.chat.completions.create(...) # OpenAI / Azure OpenAI response await client.messages.stream(...) # Anthropic response await client.models.generate_content_stream(...) # Google response await client.converse_stream(...) # Bedrock在真实源码中OpenAI 与 Anthropic 的消费路径使用AsyncExitStack进入客户端上下文确保异常路径也正确清理并在进入上下文后立即用_HttpxClient包装底层 httpx 客户端完成插桩记录请求 URL 到 OTel span见 playground_clients.pyOpenAI 流式与 L2438-L2440Anthropic 非流式。Bedrock 与 Google 则直接async with self._client_factory() as client:L1776、L2926。_HttpxClient是基于wrapt.ObjectProxy的包装器兼容 openai SDKhttpx2与其他 SDKhttpx只读取request.url写入 span 属性L3051-L3072。收益总结维度Fresh Client 模式一致性所有 Provider 完全一致资源清理通过上下文管理器自动完成凭证刷新支持 IAM 角色见下文AWS 凭证刷新附录简洁性无需包装器类插桩按请求即时应用just-in-time, per-request类型安全泛型基类确保client类型正确备选方案线程池变通Thread Pool Workaround如果 aioboto3 不可行也可以把 boto3 放到线程池中运行以释放事件循环from starlette.concurrency import run_in_threadpool # Runs boto3 in thread, doesnt block event loop await run_in_threadpool(boto3_client.converse_stream, ...)最终选择 aioboto3 的理由维度Thread Pool boto3aioboto3事件循环阻塞否在线程中运行否真正的异步线程消耗每个并发请求一个线程无复杂度同步/异步混合模式纯异步扩展性受线程池大小限制随异步事件循环扩展规格文档的结论是对大多数用户而言 boto3 的阻塞可以接受但对一个异步服务器来说aioboto3 才是正确的架构选择。技术附录各 SDK 客户端生命周期细节本节给出各 SDK 异步上下文管理器实现的详细代码引证对应各 SDK 上游源码可在各自仓库中按规格文档标注的 commit 与行号查阅。OpenAI SDKOpenAI Python SDK 基于 httpx 实现异步上下文管理器协议。其__aenter__直接返回自身__aexit__调用await self.close()async def __aenter__(self: _T) - _T: return self async def __aexit__(self, exc_type, exc, exc_tb) - None: await self.close()close()释放底层 httpx 连接async def close(self) - None: Close the underlying HTTPX client. The client will *not* be usable after this. await self._client.aclose()Anthropic SDKAnthropic SDK 与 OpenAI 共享相同的基于 httpx 的 base client 架构上下文管理器与close()的实现完全同构async def __aenter__(self: _T) - _T: return self async def __aexit__(self, exc_type, exc, exc_tb) - None: await self.close() async def close(self) - None: await self._client.aclose()Google GenAI SDKGoogle GenAI SDK 提供独立的同步/异步客户端及显式生命周期方法。AsyncClient的异步上下文管理器调用aclose()async def __aenter__(self) - AsyncClient: return self async def __aexit__(self, exc_type, exc_value, traceback) - None: await self.aclose()aclose()关闭异步 API 客户端与 nextgen 客户端async def aclose(self) - None: Closes the async client explicitly. However, it doesnt close the sync client, which can be closed using the Client.close() method or using the context manager. await self._api_client.aclose() if self._has_nextgen_client: await self._nextgen_client.close()aioboto3 / aiobotocoreAWS Bedrockaioboto3 需要完全不同的模式因为客户端创建本身是异步的并涉及凭证解析。aioboto3 Session继承 boto3替换为异步 botocoreclass Session(boto3.session.Session): def __init__(self, ...): if botocore_session is not None: self._session botocore_session else: # Create a new default session self._session aiobotocore.session.get_session()ClientCreatorContext包装异步客户端创建class ClientCreatorContext: def __init__(self, coro): self._coro coro self._client None async def __aenter__(self) - AioBaseClient: self._client await self._coro return await self._client.__aenter__() async def __aexit__(self, exc_type, exc_val, exc_tb): await self._client.__aexit__(exc_type, exc_val, exc_tb)AioBaseClient 上下文管理器管理 HTTP 会话async def __aenter__(self): await self._endpoint.http_session.__aenter__() return self async def __aexit__(self, exc_type, exc_val, exc_tb): await self._endpoint.http_session.__aexit__(exc_type, exc_val, exc_tb)HTTP Session 生命周期aiohttp connector 清理async def __aenter__(self): assert self._sessions is None self._sessions {} return self async def __aexit__(self, exc_type, exc_val, exc_tb): assert self._sessions is not None, Session was never entered self._sessions.clear() await self._exit_stack.aclose() # Make _sessions unusable once context is exited self._sessions None注意最后一行退出上下文后_sessions被置为None客户端变得不可用这正是会话生命周期绑定上下文作用域的直接证据。为什么 aioboto3 与 OpenAI/Anthropic 不同维度OpenAI/Anthropicaioboto3客户端创建同步异步凭证解析等HTTP 库httpxaiohttp连接管理内部、惰性显式、通过上下文管理器凭证处理静态 API key可能过期IAM 角色、STS tokenaioboto3 强制使用上下文管理器原因有三一是异步客户端创建凭证解析、端点发现、连接建立无法在__init__中完成二是 aiohttp 需要显式清理与可依赖垃圾回收的 httpx 不同aiohttp 连接池必须显式关闭三是凭证刷新对 IAM 角色而言新建客户端能自动获取刷新后的凭证。boto3 vs aiobotocore代码级对比botocore同步——阻塞式 socket 读取def read(self, amtNone): Read at most amt bytes from the stream. try: chunk self._raw_stream.read(amt) # ← Blocking urllib3 socket read except URLLib3ReadTimeoutError as e: raise ReadTimeoutError(endpoint_urle.url, errore) return chunkself._raw_stream.read()会阻塞整个 Python 线程从而阻塞 asyncio 事件循环直到数据从 socket 到达。boto3 Session.client()——直接返回客户端return self._session.create_client(service_name, **create_client_kwargs)boto3 的Session.client()同步立即返回客户端底层是带阻塞 urllib3 连接的同步客户端。aiobotocore Session.create_client()——包装为异步上下文管理器def create_client(self, *args, **kwargs): return ClientCreatorContext(self._create_client(*args, **kwargs))aiobotocore 把异步客户端创建包装在ClientCreatorContext中强制async with使用。真正的客户端创建在_create_client()中异步完成包括异步凭证解析。aiobotocore异步——非阻塞 awaitasync def read(self, amtNone): Read at most amt bytes from the stream. try: chunk await self.__wrapped__.content.read( amt if amt is not None else -1 ) except asyncio.TimeoutError as e: raise AioReadTimeoutError(endpoint_urlself.__wrapped__.url, errore) return chunkawait关键字把控制权交还给事件循环。底层 aiohttp 使用 asyncio futureasync def _wait(self, func_name: str) - None: waiter self._waiter self._loop.create_future() # ← Create asyncio future try: with self._timer: await waiter # ← Yield to event loop until data arrives finally: self._waiter None async def read(self, n: int -1) - bytes: while not self._buffer and not self._eof: await self._wait(read) # ← Suspends coroutine, other tasks can run return self._read_nowait(n)关键差异在于await self._wait(read)挂起协程并把控制权交还事件循环future 在数据经 socket 回调到达时被解析协程随即恢复。等待期间处理其他请求的协程可以正常执行。附录AWS 凭证刷新Credential RefreshAWS 凭证分为两类刷新行为截然不同。决策逻辑botocore 会话中的决策逻辑如下简化if aws_access_key_id is not None and aws_secret_access_key is not None: # Explicit credentials → static Credentials (no refresh) credentials botocore.credentials.Credentials( access_keyaws_access_key_id, secret_keyaws_secret_access_key, tokenaws_session_token, ) else: # No explicit creds → use credential resolver chain # This may return RefreshableCredentials for IAM roles credentials self.get_credentials()静态凭证显式传入当显式凭证传给会话时session aioboto3.Session( aws_access_key_idAKIA..., aws_secret_access_key..., aws_session_token..., # Optional, for temporary credentials )这些使用 botocore 的Credentials类仅存储值class Credentials: Holds the credentials needed to authenticate requests. def __init__(self, access_key, secret_key, tokenNone, ...): self.access_key access_key self.secret_key secret_key self.token token不会自动刷新。若凭证过期API 调用将失败并抛出ExpiredTokenException。可刷新凭证IAM 角色对于 IAM 角色EC2 实例配置文件、ECS 任务角色、Lambda 执行角色botocore 使用RefreshableCredentialsclass RefreshableCredentials(Credentials): Knows how to refresh itself. def __init__(self, ..., expiry_time, refresh_using, ...): self._refresh_using refresh_using # Callback to fetch new credentials self._expiry_time expiry_time self._advisory_refresh_timeout 15 * 60 # 15 min before expiry self._mandatory_refresh_timeout 10 * 60 # 10 min before expiry每次访问凭证时都会调用_refresh()def _refresh(self): if not self.refresh_needed(self._advisory_refresh_timeout): return # Credentials still valid, no refresh needed if self._refresh_lock.acquire(False): try: self._protected_refresh(is_mandatory...) finally: self._refresh_lock.release()_refresh_using是拉取新凭证的回调在过期前 15 分钟触发建议刷新advisory过期前 10 分钟触发强制刷新mandatory并通过锁防止并发刷新竞争。对 Phoenix 的影响自定义 Provider 配置存储的是数据库中的显式凭证见 model_provider.py 中AWSBedrockAuthenticationMethodAccessKeys与各 Provider config 的get_client_factory。这些是静态凭证不会自动刷新。如果用户配置的是会过期的临时凭证STS token必须手动更新存储的凭证。使用 IAM 角色的部署如 Phoenix 运行在带实例配置文件的 EC2/ECS 上会从环境解析凭证并自动刷新。但是自定义 Provider 工厂中的without_env_vars(AWS_*)环境变量隔离见 model_provider.py同样适用于OPENAI_*、ANTHROPIC_*、GOOGLE_*、GEMINI_*等前缀实现见 env_vars.py意味着IAM 角色凭证只对内置 Provider 生效不会被自定义配置使用——这是为了防止服务器环境中的凭证被自定义端点窃取SSRF/凭证外泄防护见_resolve_provider_api_key的注释与校验逻辑playground_clients.py。参考资料与延伸阅读规格文档 async-llm-client-lifecycle.md 的 References 部分按 commit 固定引用了 boto3、botocore、OpenAI Python SDK、Anthropic Python SDK、Google GenAI Python SDK、aioboto3、aiobotocore、aiohttp 等上游仓库的源码位置可在各自仓库按对应 commit 与行号核对本文中的代码片段。仓库内的相关实现文件包括客户端基类与各 Provider 实现playground_clients.py客户端注册表playground_registry.py工厂协议、限流键与自定义 Provider 配置model_provider.py环境变量隔离工具env_vars.py【免费下载链接】phoenixAI Observability Evaluation项目地址: https://gitcode.com/gh_mirrors/phoenix13/phoenix创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考