Phoenix Playground 異步 LLM 客戶端生命周期管理統(tǒng)一異步上下文管理器模式的設計與實現(xiàn)【免費下載鏈接】phoenixAI Observability Evaluation項目地址: https://gitcode.com/gh_mirrors/phoenix13/phoenix本文基于 Phoenix 內(nèi)部設計規(guī)格 async-llm-client-lifecycle.md深入剖析 Phoenix Playground 基礎設施中如何通過統(tǒng)一異步上下文管理器工廠模式管理 OpenAI、Azure OpenAI、Anthropic、Google GenAI、AWS Bedrock 等 LLM Provider 客戶端的完整生命周期。你將理解 boto3 為何會凍結(jié) asyncio 事件循環(huán)、aioboto3 為何強制使用async with、各 SDK 上下文管理器實現(xiàn)的差異以及 Phoenix 在源碼中如何落地這一模式——讀完即可在自己的異步 LLM 服務中正確設計客戶端創(chuàng)建、復用與資源清理策略。概述為什么需要統(tǒng)一的客戶端生命周期管理Phoenix 的 Playground模型實驗場功能支持在同一個異步服務內(nèi)調(diào)用多個 LLM ProviderOpenAI、Azure OpenAI、Anthropic、Google GenAI、AWS Bedrock每個 Provider 的 SDK 擁有截然不同的客戶端生命周期模式。規(guī)格文檔 async-llm-client-lifecycle.md 描述了在 Phoenix 全部 LLM Provider 客戶端上引入統(tǒng)一異步上下文管理器模式unified async context manager pattern的動機與設計保證 HTTP 連接生命周期的正確管理創(chuàng)建、使用、顯式關閉提供與底層 SDK 無關的一致接口修復 AWS Bedrock 的 boto3 阻塞事件循環(huán)問題。對應實現(xiàn)位于 playground_clients.py客戶端工廠的類型定義與各 Provider 的工廠實現(xiàn)位于 model_provider.py。變更內(nèi)容各 Provider 客戶端模式的演進規(guī)格文檔給出了本次設計的變更對照表Provider變更前變更后影響AWS Bedrockboto3阻塞 I/Oaioboto3異步 I/O修復事件循環(huán)阻塞OpenAI每請求新建客戶端每請求新建客戶端無變化Azure OpenAI每請求新建客戶端每請求新建客戶端無變化Anthropic每請求新建客戶端每請求新建客戶端無變化Google GenAI每請求新建客戶端每請求新建客戶端無變化核心洞察所有 Provider 原本就采用每請求新建客戶端fresh client per request的模式因此已經(jīng)承擔了每次請求約 20–250ms 的連接開銷。本次改造并未給非 Bedrock Provider 引入新的開銷——它只修復了 Bedrock 的阻塞行為。統(tǒng)一工廠模式unified factory pattern以顯式異步上下文管理器正式化了既有的 fresh-client 模式確保所有 Provider 的資源清理行為一致。問題陳述各 SDK 客戶端生命周期的差異Phoenix 的 Playground 支持多個 LLM Provider每個 Provider 的 SDK 客戶端生命周期模式各不相同OpenAI / Azure OpenAI / Anthropic客戶端可以同步創(chuàng)建但持有 HTTP 連接應顯式關閉AWS Bedrockboto3同步客戶端使用阻塞 I/O會凍結(jié)異步事件循環(huán)Google GenAI同步/異步雙客戶端生命周期管理相互獨立。值得注意的是Azure OpenAI 復用了同一個AsyncOpenAI客戶端僅通過自定義base_url指向 Azure 端點形如https://{endpoint}/openai/v1/從而避免了單獨的AsyncAzureOpenAI類。在 playground_clients.py 中可以看到 base_url 的構(gòu)造邏輯# Construct the v1 API base URL endpoint endpoint.rstrip(/) base_url (endpoint if endpoint.endswith(/openai/v1) else f{endpoint}/openai/v1) /為什么 boto3 會阻塞事件循環(huán)使用yield配合 boto3 流式輸出并不會讓它變成異步。yield關鍵字只是創(chuàng)建生成器以便增量返回數(shù)據(jù)并不改變底層 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每一次網(wǎng)絡讀取都是阻塞式 socket 操作。在這些等待期間asyncio 事件循環(huán)被凍結(jié)其他協(xié)程無法執(zhí)行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實際影響評估低影響場景阻塞可以接受本地開發(fā)、單用戶使用生產(chǎn)環(huán)境并發(fā) Bedrock 流式請求少于 5 個專用 Bedrock-only 部署且沒有其他異步工作負載。高影響場景強烈建議使用 aioboto310 個并發(fā) Bedrock 流混合異步工作負載數(shù)據(jù)庫查詢、其他 API 調(diào)用共享同一個事件循環(huán)對尾延遲tail latency敏感的延遲敏感型應用。場景boto3 影響單流約 100 tokens總計約 3-5s 阻塞用戶本來就在等待5 個并發(fā)流每個流延遲增加約 20-50%10 個并發(fā)流顯著串行化延遲疊加放大混合工作負載其他異步操作DB、HTTP在 Bedrock I/O 期間被餓死缺少生命周期管理的后果負載下連接池耗盡connection pool exhaustion長運行應用中的資源泄漏resource leaks因連接池耗盡引發(fā)的APITimeoutError尤其在流式場景下。為什么 aioboto3 要求異步上下文管理器與 OpenAI/Anthropic SDK可以同步創(chuàng)建客戶端、稍后可選關閉不同aioboto3 有更嚴格的約束這直接驅(qū)動了本次設計異步客戶端創(chuàng)建aioboto3 的session.client()返回的是ClientCreatorContext而非客戶端本身。只有進入異步上下文管理器async with之后才能拿到真正的客戶端。這是因為憑證解析、端點發(fā)現(xiàn)和 HTTP 會話建立都是異步操作。aiohttp 需要顯式清理aiohttp 的__del__對未關閉的會話只會發(fā)出ResourceWarning并不會真正關閉連接。不調(diào)用__aexit__會遺留已分配的 TCP 連接。由于 aiohttp connector 默認連接數(shù)上限為 100未關閉的客戶端最終會導致新請求等待可用連接進而引發(fā)超時。會話與客戶端的生命周期綁定在 boto3 中可以無限期持有客戶端而在 aioboto3 中客戶端的 HTTP 會話與上下文管理器作用域綁定——退出上下文會將_sessions置為None之后的 API 調(diào)用都會失敗。這些約束決定了不能簡單包裝aioboto3 來對齊 OpenAI/Anthropic 模式而是為所有 Provider 采用 aioboto3 的上下文管理器模式從而在所有地方保證一致性與正確的資源清理。設計決策統(tǒng)一工廠模式Unified Factory Pattern所有 Provider 使用一個返回異步上下文管理器的工廠可調(diào)用對象factory callable。工廠在閉包中捕獲必要的配置每次請求創(chuàng)建全新客戶端。規(guī)格文檔給出了核心骨架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]): ...這一骨架在倉庫中的真實實現(xiàn)位于 playground_clients.pyPlaygroundClient(ABC, Generic[ClientT])基類接收client_factory: ClientFactory[ClientT]、model_name、provider三個參數(shù)其中_client_factory類型為ClientFactory[ClientT]。所有子類通過register_llm_client裝飾器注冊到 Playground 注冊表playground_registry.py。工廠類型ClientFactory 與 LLMClientFactory在 model_provider.py 中定義了工廠協(xié)議及其實現(xiàn)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將創(chuàng)建客戶端的可調(diào)用對象與用于限流分桶的rate_limit_key配對。各 Provider 的限流鍵構(gòu)造函數(shù)也集中在此文件中例如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以體現(xiàn)按模型分桶的限流語義見 playground_clients.py。各 Provider 的工廠實現(xiàn)Provider工廠實現(xiàn)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中可以看到各內(nèi)置 Provider 的真實工廠構(gòu)造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)其本身實現(xiàn)了__aenter__/__aexit__天然是異步上下文管理器Google GenAIL3516-L3524由于 Google 的AsyncClient.__aexit__簽名不符合AbstractAsyncContextManager協(xié)議返回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先構(gòu)造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) )對于內(nèi)置 Provider憑證解析遵循三級優(yōu)先級請求顯式憑證 → 數(shù)據(jù)庫加密 Secret → 進程環(huán)境變量見_resolve_provider_api_keyL3185-L3217。規(guī)格文檔還特別指出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。收益總結(jié)維度Fresh Client 模式一致性所有 Provider 完全一致資源清理通過上下文管理器自動完成憑證刷新支持 IAM 角色見下文AWS 憑證刷新附錄簡潔性無需包裝器類插樁按請求即時應用just-in-time, per-request類型安全泛型基類確保client類型正確備選方案線程池變通Thread Pool Workaround如果 aioboto3 不可行也可以把 boto3 放到線程池中運行以釋放事件循環(huán)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事件循環(huán)阻塞否在線程中運行否真正的異步線程消耗每個并發(fā)請求一個線程無復雜度同步/異步混合模式純異步擴展性受線程池大小限制隨異步事件循環(huán)擴展規(guī)格文檔的結(jié)論是對大多數(shù)用戶而言 boto3 的阻塞可以接受但對一個異步服務器來說aioboto3 才是正確的架構(gòu)選擇。技術附錄各 SDK 客戶端生命周期細節(jié)本節(jié)給出各 SDK 異步上下文管理器實現(xiàn)的詳細代碼引證對應各 SDK 上游源碼可在各自倉庫中按規(guī)格文檔標注的 commit 與行號查閱。OpenAI SDKOpenAI Python SDK 基于 httpx 實現(xiàn)異步上下文管理器協(xié)議。其__aenter__直接返回自身__aexit__調(diào)用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 架構(gòu)上下文管理器與close()的實現(xiàn)完全同構(gòu)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的異步上下文管理器調(diào)用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 需要完全不同的模式因為客戶端創(chuàng)建本身是異步的并涉及憑證解析。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包裝異步客戶端創(chuàng)建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客戶端變得不可用這正是會話生命周期綁定上下文作用域的直接證據(jù)。為什么 aioboto3 與 OpenAI/Anthropic 不同維度OpenAI/Anthropicaioboto3客戶端創(chuàng)建同步異步憑證解析等HTTP 庫httpxaiohttp連接管理內(nèi)部、惰性顯式、通過上下文管理器憑證處理靜態(tài) API key可能過期IAM 角色、STS tokenaioboto3 強制使用上下文管理器原因有三一是異步客戶端創(chuàng)建憑證解析、端點發(fā)現(xiàn)、連接建立無法在__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 事件循環(huán)直到數(shù)據(jù)從 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 把異步客戶端創(chuàng)建包裝在ClientCreatorContext中強制async with使用。真正的客戶端創(chuàng)建在_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關鍵字把控制權交還給事件循環(huán)。底層 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)掛起協(xié)程并把控制權交還事件循環(huán)future 在數(shù)據(jù)經(jīng) socket 回調(diào)到達時被解析協(xié)程隨即恢復。等待期間處理其他請求的協(xié)程可以正常執(zhí)行。附錄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()靜態(tài)憑證顯式傳入當顯式憑證傳給會話時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 調(diào)用將失敗并拋出ExpiredTokenException?？伤⑿聭{證IAM 角色對于 IAM 角色EC2 實例配置文件、ECS 任務角色、Lambda 執(zhí)行角色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每次訪問憑證時都會調(diào)用_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是拉取新憑證的回調(diào)在過期前 15 分鐘觸發(fā)建議刷新advisory過期前 10 分鐘觸發(fā)強制刷新mandatory并通過鎖防止并發(fā)刷新競爭。對 Phoenix 的影響自定義 Provider 配置存儲的是數(shù)據(jù)庫中的顯式憑證見 model_provider.py 中AWSBedrockAuthenticationMethodAccessKeys與各 Provider config 的get_client_factory。這些是靜態(tài)憑證不會自動刷新。如果用戶配置的是會過期的臨時憑證STS token必須手動更新存儲的憑證。使用 IAM 角色的部署如 Phoenix 運行在帶實例配置文件的 EC2/ECS 上會從環(huán)境解析憑證并自動刷新。但是自定義 Provider 工廠中的without_env_vars(AWS_*)環(huán)境變量隔離見 model_provider.py同樣適用于OPENAI_*、ANTHROPIC_*、GOOGLE_*、GEMINI_*等前綴實現(xiàn)見 env_vars.py意味著IAM 角色憑證只對內(nèi)置 Provider 生效不會被自定義配置使用——這是為了防止服務器環(huán)境中的憑證被自定義端點竊取SSRF/憑證外泄防護見_resolve_provider_api_key的注釋與校驗邏輯playground_clients.py。參考資料與延伸閱讀規(guī)格文檔 async-llm-client-lifecycle.md 的 References 部分按 commit 固定引用了 boto3、botocore、OpenAI Python SDK、Anthropic Python SDK、Google GenAI Python SDK、aioboto3、aiobotocore、aiohttp 等上游倉庫的源碼位置可在各自倉庫按對應 commit 與行號核對本文中的代碼片段。倉庫內(nèi)的相關實現(xiàn)文件包括客戶端基類與各 Provider 實現(xiàn)playground_clients.py客戶端注冊表playground_registry.py工廠協(xié)議、限流鍵與自定義 Provider 配置model_provider.py環(huán)境變量隔離工具env_vars.py【免費下載鏈接】phoenixAI Observability Evaluation項目地址: https://gitcode.com/gh_mirrors/phoenix13/phoenix創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考