Merge pull request #2068 from BloggerBust/fix/async-gather-reuse-hang

fix(async,ci): prevent coroutine re-await & stabilize async evals

Author: penguine-ip

.github/workflows/test_core.yml

@@ -8,6 +8,10 @@ on:
 jobs:
   test:
     runs-on: ubuntu-latest
+    env:
+      # Expose once at job level because forked PRs can't use secrets.* in `if:` conditions.
+      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
+
     steps:
       #----------------------------------------------
       #       check-out repo and set-up python
@@ -54,20 +58,46 @@ jobs:
       #----------------------------------------------
       #              run test suite
       #----------------------------------------------
+
+      # Run tests (with secrets): full suite
       - name: Run tests
-        env:
-          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
+        if: ${{ env.OPENAI_API_KEY != '' }}
+        run: |
+          poetry run pytest -vv -rA --maxfail=1 --capture=tee-sys \
+            tests/test_core/ \
+            --ignore=tests/test_core/test_synthesizer/ \
+            --ignore=tests/test_core/test_datasets/
+
+      # Run tests (no secrets): skip e2e that require API keys
+      - name: Run tests (no secrets)
+        if: ${{ env.OPENAI_API_KEY == '' }}
         run: |
-          poetry run pytest -vv -rA --maxfail=1 --capture=tee-sys tests/test_core/ --ignore=tests/test_core/test_synthesizer/ --ignore=tests/test_core/test_datasets/ --ignore=tests/test_core/test_tracing/test_dataset_iterator.py
+          poetry run pytest -vv -rA --maxfail=1 --capture=tee-sys tests/test_core/  \
+          --ignore=tests/test_core/test_synthesizer/                                \
+          --ignore=tests/test_core/test_datasets/                                   \
+          --ignore=tests/test_core/test_tracing/test_dataset_iterator.py            \
+          --ignore=tests/test_core/test_evaluation/test_end_to_end/test_configs.py
 
       #----------------------------------------------
       #    install dev dependencies (including chromadb) and run synthesizer tests
       #----------------------------------------------
       - name: Install dev dependencies
         run: poetry install --no-interaction --with dev
 
-      - name: Run core tests with dev dependencies
-        env:
-          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
+      # Dev tests (with secrets)
+      - name: Run dev tests
+        if: ${{ env.OPENAI_API_KEY != '' }}
+        run: |
+          poetry run pytest -vv -rA --maxfail=1 --capture=tee-sys -o faulthandler_timeout=300 \
+          tests/test_core/test_synthesizer/ tests/test_core/test_datasets/
+
+      # Dev tests (no secrets)
+      - name: Run dev tests (no secrets)
+        if: ${{ env.OPENAI_API_KEY == '' }}
         run: |
-          poetry run pytest -vv -rA --maxfail=1 --capture=tee-sys tests/test_core/test_synthesizer/ tests/test_core/test_datasets/ --ignore=tests/test_core/test_tracing/test_dataset_iterator.py
+          poetry run pytest -vv -rA --maxfail=1 --capture=tee-sys tests/test_core/test_synthesizer/ tests/test_core/test_datasets/ \
+          --ignore=tests/test_core/test_tracing/test_dataset_iterator.py            \
+          --ignore=tests/test_core/test_synthesizer/test_context_generator.py       \
+          --ignore=tests/test_core/test_synthesizer/test_conversation_simulator.py  \
+          --ignore=tests/test_core/test_synthesizer/test_generate_from_goldens.py   \
+          --ignore=tests/test_core/test_synthesizer/test_synthesizer.py

deepeval/config/settings.py

@@ -281,6 +281,7 @@ class Settings(BaseSettings):
     #
     # Telemetry and Debug
     #
+    DEEPEVAL_DEBUG_ASYNC: Optional[bool] = None
     DEEPEVAL_TELEMETRY_OPT_OUT: Optional[bool] = None
     DEEPEVAL_UPDATE_WARNING_OPT_IN: Optional[bool] = None
     DEEPEVAL_GRPC_LOGGING: Optional[bool] = None
@@ -303,6 +304,19 @@ class Settings(BaseSettings):
     MEDIA_IMAGE_CONNECT_TIMEOUT_SECONDS: float = 3.05
     MEDIA_IMAGE_READ_TIMEOUT_SECONDS: float = 10.0
 
+    #
+    # Async Task Configuration
+    #
+
+    # Maximum time allowed for a single task to complete
+    DEEPEVAL_PER_TASK_TIMEOUT_SECONDS: int = (
+        300  # Set to float('inf') to disable timeout
+    )
+
+    # Buffer time for gathering results from all tasks, added to the longest task duration
+    # Increase if many tasks are running concurrently
+    DEEPEVAL_TASK_GATHER_BUFFER_SECONDS: int = 60
+
     ##############
     # Validators #
     ##############

deepeval/dataset/dataset.py

@@ -1,13 +1,12 @@
 from asyncio import Task
-from typing import Iterator, List, Optional, Union, Literal
+from typing import TYPE_CHECKING, Iterator, List, Optional, Union, Literal
 from dataclasses import dataclass, field
 from opentelemetry.trace import Tracer
 from opentelemetry.context import Context, attach, detach
 from rich.console import Console
 from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn
 import json
 import csv
-import webbrowser
 import os
 import datetime
 import time
@@ -17,6 +16,7 @@
 
 from deepeval.confident.api import Api, Endpoints, HttpMethods
 from deepeval.dataset.utils import (
+    coerce_to_task,
     convert_test_cases_to_goldens,
     convert_goldens_to_test_cases,
     convert_convo_goldens_to_convo_test_cases,
@@ -49,11 +49,18 @@
 from deepeval.test_run import (
     global_test_run_manager,
 )
-from deepeval.dataset.types import global_evaluation_tasks
 from deepeval.openai.utils import openai_test_case_pairs
 from deepeval.tracing import trace_manager
 from deepeval.tracing.tracing import EVAL_DUMMY_SPAN_NAME
 
+if TYPE_CHECKING:
+    from deepeval.evaluate.configs import (
+        AsyncConfig,
+        DisplayConfig,
+        CacheConfig,
+        ErrorConfig,
+    )
+
 
 valid_file_types = ["csv", "json", "jsonl"]
 
@@ -1230,7 +1237,7 @@ def evals_iterator(
                 )
 
     def evaluate(self, task: Task):
-        global_evaluation_tasks.append(task)
+        coerce_to_task(task)
 
     def _start_otel_test_run(self, tracer: Optional[Tracer] = None) -> Context:
         _tracer = check_tracer(tracer)

deepeval/dataset/types.py

@@ -1,17 +1,25 @@
-class EvaluationTasks:
-    tasks: list = []
+import asyncio
 
-    def append(self, t):
-        self.tasks.append(t)
+from typing import Any
+from deepeval.dataset.utils import coerce_to_task
 
-    def get_tasks(self):
-        return self.tasks
 
-    def num_tasks(self):
-        return len(self.tasks)
+class EvaluationTasks:
 
-    def clear_tasks(self):
-        self.tasks.clear()
+    def __init__(self):
+        self._tasks: list[asyncio.Future] = []
 
+    def append(self, obj: Any):
+        self._tasks.append(coerce_to_task(obj))
+
+    def get_tasks(self) -> list[asyncio.Future]:
+        return list(self._tasks)
+
+    def num_tasks(self):
+        return len(self._tasks)
 
-global_evaluation_tasks = EvaluationTasks()
+    def clear_tasks(self) -> None:
+        for t in self._tasks:
+            if not t.done():
+                t.cancel()
+        self._tasks.clear()

deepeval/dataset/utils.py

@@ -1,10 +1,10 @@
-from typing import List, Optional, Any
+import asyncio
+import inspect
 import json
 import re
 
+from typing import List, Optional, Any
 from opentelemetry.trace import Tracer
-from opentelemetry import trace
-from opentelemetry.trace import NoOpTracerProvider
 
 from deepeval.dataset.api import Golden
 from deepeval.dataset.golden import ConversationalGolden
@@ -174,3 +174,31 @@ def check_tracer(tracer: Optional[Tracer] = None) -> Tracer:
         )
 
     return GLOBAL_TEST_RUN_TRACER
+
+
+def coerce_to_task(obj: Any) -> asyncio.Future[Any]:
+    # already a Task so just return it
+    if isinstance(obj, asyncio.Task):
+        return obj
+
+    # If it is a future, it is already scheduled, so just return it
+    if asyncio.isfuture(obj):
+        # type: ignore[return-value]  # it is an awaitable, gather accepts it
+        return obj
+
+    # bare coroutine must be explicitly scheduled using create_task to bind to loop & track
+    if asyncio.iscoroutine(obj):
+        return asyncio.create_task(obj)
+
+    # generic awaitable (any object with __await__) will need to be wrapped so create_task accepts it
+    if inspect.isawaitable(obj):
+
+        async def _wrap(awaitable):
+            return await awaitable
+
+        return asyncio.create_task(_wrap(obj))
+
+    # not awaitable, so time to sound the alarm!
+    raise TypeError(
+        f"Expected Task/Future/coroutine/awaitable, got {type(obj).__name__}"
+    )