infra: drop py3.8 CI

.github/workflows/python_test.yml

@@ -1,4 +1,4 @@
 name: Python CI
 
 on:
   push:
@@ -19,11 +19,11 @@
     strategy:
       matrix:
         python-version:
-          - "3.8"
           - "3.9"
           - "3.10"
           - "3.11"
           - "3.12"
+          - "3.13"
     defaults:
       run:
         working-directory: python

python/langsmith/_internal/_aiter.py

@@ -334,25 +334,6 @@ def accepts_context(callable: Callable[..., Any]) -> bool:
         return False
 
 
-# Ported from Python 3.9+ to support Python 3.8
-async def aio_to_thread(
-    func, /, *args, __ctx: Optional[contextvars.Context] = None, **kwargs
-):
-    """Asynchronously run function *func* in a separate thread.
-
-    Any *args and **kwargs supplied for this function are directly passed
-    to *func*. Also, the current :class:`contextvars.Context` is propagated,
-    allowing context variables from the main thread to be accessed in the
-    separate thread.
-
-    Return a coroutine that can be awaited to get the eventual result of *func*.
-    """
-    loop = asyncio.get_running_loop()
-    ctx = __ctx or contextvars.copy_context()
-    func_call = functools.partial(ctx.run, func, *args, **kwargs)
-    return await loop.run_in_executor(None, func_call)
-
-
 @functools.lru_cache(maxsize=1)
 def asyncio_accepts_context():
     """Check if the current asyncio event loop accepts a context argument."""

python/langsmith/evaluation/_arunner.py

@@ -344,13 +344,13 @@ async def aevaluate_existing(
     project = (
         experiment
         if isinstance(experiment, schemas.TracerSession)
-        else (await aitertools.aio_to_thread(_load_experiment, experiment, client))
+        else (await asyncio.to_thread(_load_experiment, experiment, client))
     )
-    runs = await aitertools.aio_to_thread(
+    runs = await asyncio.to_thread(
         _load_traces, experiment, client, load_nested=load_nested
     )
-    data_map = await aitertools.aio_to_thread(_load_examples_map, client, project)
-    data = [data_map[run.reference_example_id] for run in runs]
+    data_map = await asyncio.to_thread(_load_examples_map, client, project)
+    data = [data_map[cast(uuid.UUID, run.reference_example_id)] for run in runs]
     return await _aevaluate(
         runs,
         data=data,
@@ -384,7 +384,7 @@ async def _aevaluate(
     )
     client = client or rt.get_cached_client()
     runs = None if is_async_target else cast(Iterable[schemas.Run], target)
-    experiment_, runs = await aitertools.aio_to_thread(
+    experiment_, runs = await asyncio.to_thread(
         _resolve_experiment,
         experiment,
         runs,
@@ -770,7 +770,7 @@ async def _aapply_summary_evaluators(
                     for result in flattened_results:
                         feedback = result.dict(exclude={"target_run_id"})
                         evaluator_info = feedback.pop("evaluator_info", None)
-                        await aitertools.aio_to_thread(
+                        await asyncio.to_thread(
                             self.client.create_feedback,
                             **feedback,
                             run_id=None,

python/langsmith/run_helpers.py

@@ -487,7 +487,7 @@ async def async_wrapper(
             **kwargs: Any,
         ) -> Any:
             """Async version of wrapper function."""
-            run_container = await aitertools.aio_to_thread(
+            run_container = await asyncio.to_thread(
                 _setup_run,
                 func,
                 container_input=container_input,
@@ -516,19 +516,17 @@ async def async_wrapper(
             except BaseException as e:
                 # shield from cancellation, given we're catching all exceptions
                 await asyncio.shield(
-                    aitertools.aio_to_thread(_on_run_end, run_container, error=e)
+                    asyncio.to_thread(_on_run_end, run_container, error=e)
                 )
                 raise e
-            await aitertools.aio_to_thread(
-                _on_run_end, run_container, outputs=function_result
-            )
+            await asyncio.to_thread(_on_run_end, run_container, outputs=function_result)
             return function_result
 
         @functools.wraps(func)
         async def async_generator_wrapper(
             *args: Any, langsmith_extra: Optional[LangSmithExtra] = None, **kwargs: Any
         ) -> AsyncGenerator:
-            run_container = await aitertools.aio_to_thread(
+            run_container = await asyncio.to_thread(
                 _setup_run,
                 func,
                 container_input=container_input,
@@ -574,15 +572,15 @@ async def async_generator_wrapper(
                     yield item
             except BaseException as e:
                 await asyncio.shield(
-                    aitertools.aio_to_thread(
+                    asyncio.to_thread(
                         _on_run_end,
                         run_container,
                         error=e,
                         outputs=_get_function_result(results, reduce_fn),
                     )
                 )
                 raise e
-            await aitertools.aio_to_thread(
+            await asyncio.to_thread(
                 _on_run_end,
                 run_container,
                 outputs=_get_function_result(results, reduce_fn),
@@ -700,7 +698,7 @@ def stream_wrapper(
         async def async_stream_wrapper(
             *args: Any, langsmith_extra: Optional[LangSmithExtra] = None, **kwargs: Any
         ) -> Any:
-            trace_container = await aitertools.aio_to_thread(
+            trace_container = await asyncio.to_thread(
                 _setup_run,
                 func,
                 container_input=container_input,
@@ -716,7 +714,7 @@ async def async_stream_wrapper(
                     kwargs.pop("config", None)
                 stream = await func(*args, **kwargs)
             except Exception as e:
-                await aitertools.aio_to_thread(_on_run_end, trace_container, error=e)
+                await asyncio.to_thread(_on_run_end, trace_container, error=e)
                 raise
 
             if hasattr(stream, "__aiter__"):
@@ -726,7 +724,7 @@ async def async_stream_wrapper(
                 return _TracedStream(stream, trace_container, reduce_fn)
 
             # If it's not iterable, end the trace immediately
-            await aitertools.aio_to_thread(_on_run_end, trace_container, outputs=stream)
+            await asyncio.to_thread(_on_run_end, trace_container, outputs=stream)
             return stream
 
         if inspect.isasyncgenfunction(func):
@@ -1013,7 +1011,7 @@ async def __aenter__(self) -> run_trees.RunTree:
             run_trees.RunTree: The newly created run.
         """
         ctx = copy_context()
-        result = await aitertools.aio_to_thread(self._setup, __ctx=ctx)
+        result = await asyncio.to_thread(self._setup)
         # Set the context for the current thread
         _set_tracing_context(get_tracing_context(ctx))
         return result
@@ -1034,14 +1032,10 @@ async def __aexit__(
         ctx = copy_context()
         if exc_type is not None:
             await asyncio.shield(
-                aitertools.aio_to_thread(
-                    self._teardown, exc_type, exc_value, traceback, __ctx=ctx
-                )
+                asyncio.to_thread(self._teardown, exc_type, exc_value, traceback)
             )
         else:
-            await aitertools.aio_to_thread(
-                self._teardown, exc_type, exc_value, traceback, __ctx=ctx
-            )
+            await asyncio.to_thread(self._teardown, exc_type, exc_value, traceback)
         _set_tracing_context(get_tracing_context(ctx))
 
 
@@ -1712,9 +1706,7 @@ def __init__(
 
     async def _aend_trace(self, error: Optional[BaseException] = None):
         ctx = copy_context()
-        await asyncio.shield(
-            aitertools.aio_to_thread(self._end_trace, error, __ctx=ctx)
-        )
+        await asyncio.shield(asyncio.to_thread(self._end_trace, error))
         _set_tracing_context(get_tracing_context(ctx))
 
     async def __anext__(self) -> T: