jdZddlmZddlmZmZmZmZerddlm Z edZ GddZ e a ddd Z ddZddZdS)zRuntime overrides for LangSmith. This module provides hooks to override LangSmith's default runtime behavior, primarily for environments with constrained async runtimes (e.g., Temporal). ) annotations) TYPE_CHECKINGAnyCallableOptional) Awaitable).zAwaitable[Any]c"eZdZdZdZ dddZdS) RuntimeOverridesaOverrides for LangSmith runtime behavior. This class allows overriding default async implementations for environments that don't support certain asyncio features (e.g., Temporal doesn't support ``run_in_executor``). Example: import langsmith import contextvars async def my_aio_to_thread( default_aio_to_thread, ctx, func, /, *args, **kwargs ): # Custom implementation return ctx.run(func, *args, **kwargs) langsmith.set_runtime_overrides(aio_to_thread=my_aio_to_thread) # Reset to defaults langsmith.set_runtime_overrides()  aio_to_threadNr Optional[AioToThread]c||_dS)aInitialize runtime overrides. Args: aio_to_thread: Custom async-to-thread implementation, with signature ``async def (default_aio_to_thread, ctx, func, /, *args, **kwargs)``. ``default_aio_to_thread`` is LangSmith's default implementation, which the override can call to fall back to default behavior (e.g., when outside a constrained runtime context). ``ctx`` is the ``contextvars.Context`` LangSmith wants ``func`` to run inside; tracing state will be read back from this Context after the call. Override for runtimes like Temporal that don't support ``asyncio.run_in_executor``. Nr )selfr s f/lsinfo/ai/hellotax_ai/base_platform/venv/lib/python3.11/site-packages/langsmith/_runtime_overrides.py__init__zRuntimeOverrides.__init__0s"+N)r r )__name__ __module__ __qualname____doc__ __slots__rrrr r sC0#I04+++++++rr Nr r returnNonec&t|adS)aSet LangSmith runtime overrides. This allows customizing LangSmith's async runtime behavior for environments with constrained async runtimes (e.g., Temporal, which doesn't support ``run_in_executor``). Args: aio_to_thread: Custom async function to run sync functions asynchronously. Should have signature: ``async def aio_to_thread( default_aio_to_thread, ctx, func, /, *args, **kwargs )``. ``default_aio_to_thread`` is LangSmith's default implementation, which the override can call to fall back to default behavior. The implementation must invoke ``func`` inside ``ctx`` (e.g. ``ctx.run(func, *args, **kwargs)``) so that LangSmith's tracing state, which is read back from ``ctx`` after the call, is visible to downstream code. Pass ``None`` to use the default implementation. Example: For Temporal or similar runtimes: ```python import langsmith async def temporal_aio_to_thread( default_aio_to_thread, ctx, func, /, *args, **kwargs ): # Use the default implementation when not in a workflow if not temporalio.workflow.in_workflow(): return await default_aio_to_thread(ctx, func, *args, **kwargs) with temporalio.workflow.unsafe.sandbox_unrestricted(): return ctx.run(func, *args, **kwargs) langsmith.set_runtime_overrides(aio_to_thread=temporal_aio_to_thread) ``` Reset to defaults: ```python langsmith.set_runtime_overrides() ``` r N)r _runtime_overridesr s rset_runtime_overridesrGsb* FFFrctS)z"Get the current runtime overrides.)rrrrget_runtime_overridesr {s rboolctjduS)aReturn True iff an ``aio_to_thread`` override is currently installed. Callers use this to select a loop-behavior-independent path for re-entering a LangSmith-mutated Context (the explicit ``tracing_context(**get_tracing_context(ctx))`` fallback), rather than relying on ``asyncio.create_task(coro, context=ctx)`` which some custom event loops silently ignore. N)rr rrr_aio_to_thread_override_activer#s  +4 77rr)r r rr)rr )rr!)r __future__rtypingrrrrcollections.abcr AioToThreadr rrr r#rrrr(s #"""""999999999999*)))))) ,+,+,+,+,+,+,+,+^&%'',01G1G1G1G1Gh 8 8 8 8 8 8r