wehub-resource-sync
1023 lines
41 KiB
Python
1023 lines
41 KiB
Python
import logging
|
||
|
||
from rich.progress import (
|
||
Progress,
|
||
TextColumn,
|
||
BarColumn,
|
||
TimeElapsedColumn,
|
||
TaskProgressColumn,
|
||
)
|
||
from typing import (
|
||
Callable,
|
||
List,
|
||
Optional,
|
||
Awaitable,
|
||
Iterator,
|
||
)
|
||
import asyncio
|
||
import time
|
||
|
||
from deepeval.evaluate.configs import (
|
||
ErrorConfig,
|
||
DisplayConfig,
|
||
CacheConfig,
|
||
AsyncConfig,
|
||
)
|
||
from deepeval.tracing.tracing import (
|
||
Observer,
|
||
trace_manager,
|
||
Trace,
|
||
BaseSpan,
|
||
AgentSpan,
|
||
LlmSpan,
|
||
RetrieverSpan,
|
||
ToolSpan,
|
||
)
|
||
from deepeval.tracing.context import current_trace_context
|
||
from deepeval.tracing.api import (
|
||
BaseApiSpan,
|
||
)
|
||
from deepeval.dataset import Golden
|
||
from deepeval.contextvars import set_current_golden, reset_current_golden
|
||
from deepeval.constants import PYTEST_TRACE_TEST_WRAPPER_SPAN_NAME
|
||
from deepeval.errors import DeepEvalError
|
||
from deepeval.metrics.utils import copy_metrics
|
||
from deepeval.utils import (
|
||
shorten,
|
||
len_medium,
|
||
format_error_text,
|
||
get_per_task_timeout_seconds,
|
||
get_gather_timeout,
|
||
)
|
||
from deepeval.telemetry import capture_evaluation_run
|
||
from deepeval.metrics import BaseMetric
|
||
|
||
from deepeval.test_case import (
|
||
LLMTestCase,
|
||
)
|
||
from deepeval.test_case.api import create_api_test_case
|
||
from deepeval.test_run import (
|
||
global_test_run_manager,
|
||
TestRunManager,
|
||
)
|
||
from deepeval.evaluate.types import TestResult
|
||
from deepeval.evaluate.utils import (
|
||
create_api_trace,
|
||
create_metric_data,
|
||
create_test_result,
|
||
count_metrics_in_trace,
|
||
count_total_metrics_for_trace,
|
||
count_metrics_in_span_subtree,
|
||
extract_trace_test_results,
|
||
)
|
||
from deepeval.utils import add_pbar, update_pbar, custom_console
|
||
from deepeval.tracing.types import (
|
||
EvalMode,
|
||
EvalSession,
|
||
TestCaseMetricPair,
|
||
TraceSpanStatus,
|
||
)
|
||
from deepeval.tracing.api import TraceSpanApiStatus
|
||
from deepeval.config.settings import get_settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
from deepeval.evaluate.execute._common import (
|
||
_await_with_outer_deadline,
|
||
_execute_metric,
|
||
_log_gather_timeout,
|
||
_pick_root_for_marking,
|
||
_resolve_trace_and_root_for_task,
|
||
_skip_metrics_for_error,
|
||
_snapshot_tasks,
|
||
_trace_error,
|
||
filter_duplicate_results,
|
||
log_prompt,
|
||
)
|
||
from deepeval.evaluate.execute.agentic import (
|
||
_a_execute_agentic_test_case,
|
||
)
|
||
from deepeval.evaluate.execute.e2e import _evaluate_test_case_pairs
|
||
|
||
|
||
def _span_subtree_has_metrics(span: BaseSpan) -> bool:
|
||
"""True if ``span`` or any of its descendants declares a metric source."""
|
||
if span.metrics:
|
||
return True
|
||
return any(_span_subtree_has_metrics(c) for c in span.children)
|
||
|
||
|
||
def _has_any_evaluable_metrics(
|
||
trace_metrics: Optional[List[BaseMetric]],
|
||
traces: List[Trace],
|
||
test_case_metrics: List[TestCaseMetricPair],
|
||
) -> bool:
|
||
"""Return True if at least one metric source exists for this eval run.
|
||
|
||
Metrics can come from: ``trace_metrics`` (iterator arg), ``trace.metrics``
|
||
(``update_current_trace``/root ``@observe``), ``span.metrics`` anywhere in
|
||
a trace subtree, or ``test_case_metrics`` (external SDK path). This check
|
||
is intentionally lazy (post-iteration) since span metrics only exist after
|
||
user code has run.
|
||
"""
|
||
if trace_metrics:
|
||
return True
|
||
if test_case_metrics:
|
||
return True
|
||
for trace in traces:
|
||
if not isinstance(trace, Trace):
|
||
continue
|
||
if trace.metrics:
|
||
return True
|
||
if any(_span_subtree_has_metrics(s) for s in trace.root_spans):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _raise_no_metrics_error() -> None:
|
||
"""Raise a uniform NoMetricsError with actionable guidance."""
|
||
from deepeval.errors import NoMetricsError
|
||
|
||
raise NoMetricsError(
|
||
"evals_iterator was started but no metrics were declared anywhere. "
|
||
"An evaluation run with zero metric sources cannot produce results.\n"
|
||
)
|
||
|
||
|
||
def execute_agentic_test_cases_from_loop(
|
||
goldens: List[Golden],
|
||
trace_metrics: Optional[List[BaseMetric]],
|
||
test_results: List[TestResult],
|
||
display_config: Optional[DisplayConfig] = DisplayConfig(),
|
||
cache_config: Optional[CacheConfig] = CacheConfig(),
|
||
error_config: Optional[ErrorConfig] = ErrorConfig(),
|
||
identifier: Optional[str] = None,
|
||
_use_bar_indicator: bool = True,
|
||
_is_assert_test: bool = False,
|
||
) -> Iterator[TestResult]:
|
||
|
||
test_run_manager = global_test_run_manager
|
||
test_run_manager.save_to_disk = cache_config.write_cache
|
||
test_run_manager.get_test_run(identifier=identifier)
|
||
|
||
local_trace_manager = trace_manager
|
||
local_trace_manager.eval_session = EvalSession(mode=EvalMode.ITERATOR_SYNC)
|
||
|
||
def evaluate_test_cases(
|
||
progress: Optional[Progress] = None,
|
||
pbar_id: Optional[int] = None,
|
||
) -> Iterator[Golden]:
|
||
count = 0
|
||
show_metric_indicator = (
|
||
display_config.show_indicator and not _use_bar_indicator
|
||
)
|
||
# Per-run buffer of traces produced by user code. Accumulated locally
|
||
# so the post-iteration "any metrics?" guard only inspects THIS run.
|
||
processed_traces: List[Trace] = []
|
||
|
||
for golden in goldens:
|
||
token = set_current_golden(golden)
|
||
with capture_evaluation_run("golden"):
|
||
# yield golden
|
||
count += 1
|
||
pbar_tags_id = add_pbar(
|
||
progress, f"\t⚡ Invoking observed callback (#{count})"
|
||
)
|
||
with Observer(
|
||
"custom",
|
||
func_name=PYTEST_TRACE_TEST_WRAPPER_SPAN_NAME,
|
||
_progress=progress,
|
||
_pbar_callback_id=pbar_tags_id,
|
||
):
|
||
try:
|
||
# yield golden to user code
|
||
yield golden
|
||
# control has returned from user code without error, capture trace now
|
||
current_trace: Trace = current_trace_context.get()
|
||
processed_traces.append(current_trace)
|
||
finally:
|
||
# after user code returns control, always reset the context
|
||
reset_current_golden(token)
|
||
|
||
update_pbar(progress, pbar_tags_id)
|
||
update_pbar(progress, pbar_id)
|
||
|
||
# Create empty trace api for llm api test case
|
||
trace_api = create_api_trace(trace=current_trace, golden=golden)
|
||
|
||
# Format golden as test case to create llm api test case
|
||
test_case = LLMTestCase(
|
||
input=golden.input,
|
||
actual_output=(
|
||
str(current_trace.output)
|
||
if current_trace.output is not None
|
||
else None
|
||
),
|
||
expected_output=current_trace.expected_output,
|
||
context=current_trace.context,
|
||
retrieval_context=current_trace.retrieval_context,
|
||
metadata=golden.additional_metadata,
|
||
tools_called=current_trace.tools_called,
|
||
expected_tools=current_trace.expected_tools,
|
||
comments=golden.comments,
|
||
name=golden.name,
|
||
_dataset_alias=golden._dataset_alias,
|
||
_dataset_id=golden._dataset_id,
|
||
)
|
||
api_test_case = create_api_test_case(
|
||
test_case=test_case,
|
||
trace=trace_api,
|
||
index=count if not _is_assert_test else None,
|
||
)
|
||
|
||
# Run DFS to calculate metrics synchronously
|
||
def dfs(
|
||
span: BaseSpan,
|
||
progress: Optional[Progress] = None,
|
||
pbar_eval_id: Optional[int] = None,
|
||
):
|
||
# Create API Span
|
||
metrics: List[BaseMetric] = list(span.metrics or [])
|
||
|
||
api_span: BaseApiSpan = (
|
||
trace_manager._convert_span_to_api_span(span)
|
||
)
|
||
|
||
if isinstance(span, AgentSpan):
|
||
trace_api.agent_spans.append(api_span)
|
||
elif isinstance(span, LlmSpan):
|
||
trace_api.llm_spans.append(api_span)
|
||
log_prompt(span, test_run_manager)
|
||
elif isinstance(span, RetrieverSpan):
|
||
trace_api.retriever_spans.append(api_span)
|
||
elif isinstance(span, ToolSpan):
|
||
trace_api.tool_spans.append(api_span)
|
||
else:
|
||
trace_api.base_spans.append(api_span)
|
||
|
||
# Skip errored trace/span
|
||
if _skip_metrics_for_error(span=span, trace=current_trace):
|
||
api_span.status = TraceSpanApiStatus.ERRORED
|
||
api_span.error = span.error or _trace_error(
|
||
current_trace
|
||
)
|
||
if progress and pbar_eval_id is not None:
|
||
update_pbar(
|
||
progress,
|
||
pbar_eval_id,
|
||
advance=count_metrics_in_span_subtree(span),
|
||
)
|
||
return
|
||
|
||
for child in span.children:
|
||
dfs(child, progress, pbar_eval_id)
|
||
|
||
if not span.metrics:
|
||
return
|
||
|
||
requires_trace = any(
|
||
metric.requires_trace for metric in metrics
|
||
)
|
||
|
||
llm_test_case = None
|
||
if span.input is not None:
|
||
llm_test_case = LLMTestCase(
|
||
input=str(span.input),
|
||
actual_output=(
|
||
str(span.output)
|
||
if span.output is not None
|
||
else None
|
||
),
|
||
expected_output=span.expected_output,
|
||
context=span.context,
|
||
retrieval_context=span.retrieval_context,
|
||
tools_called=span.tools_called,
|
||
expected_tools=span.expected_tools,
|
||
)
|
||
|
||
if requires_trace:
|
||
if llm_test_case is None:
|
||
llm_test_case = LLMTestCase(input="None")
|
||
llm_test_case._trace_dict = (
|
||
trace_manager.create_nested_spans_dict(span)
|
||
)
|
||
else:
|
||
if llm_test_case is None:
|
||
api_span.status = TraceSpanApiStatus.ERRORED
|
||
api_span.error = format_error_text(
|
||
DeepEvalError(
|
||
"Span has metrics but no LLMTestCase. "
|
||
"Are you sure you called `update_current_span()`?"
|
||
)
|
||
)
|
||
if progress and pbar_eval_id is not None:
|
||
update_pbar(
|
||
progress,
|
||
pbar_eval_id,
|
||
advance=count_metrics_in_span_subtree(span),
|
||
)
|
||
return
|
||
|
||
# Preparing metric calculation
|
||
api_span.metrics_data = []
|
||
for metric in metrics:
|
||
metric.skipped = False
|
||
metric.error = None
|
||
if display_config.verbose_mode is not None:
|
||
metric.verbose_mode = display_config.verbose_mode
|
||
|
||
# Metric calculation
|
||
for metric in metrics:
|
||
metric_data = None
|
||
res = _execute_metric(
|
||
metric=metric,
|
||
test_case=llm_test_case,
|
||
show_metric_indicator=show_metric_indicator,
|
||
in_component=True,
|
||
error_config=error_config,
|
||
)
|
||
if res == "skip":
|
||
continue
|
||
|
||
metric_data = create_metric_data(metric)
|
||
api_span.metrics_data.append(metric_data)
|
||
api_test_case.update_status(metric_data.success)
|
||
update_pbar(progress, pbar_eval_id)
|
||
|
||
if trace_metrics:
|
||
current_trace.metrics = trace_metrics
|
||
|
||
trace_level_metrics_count = (
|
||
len(current_trace.metrics) if current_trace.metrics else 0
|
||
)
|
||
pbar_eval_id = add_pbar(
|
||
progress,
|
||
f" 🎯 Evaluating component(s) (#{count})",
|
||
total=count_metrics_in_trace(trace=current_trace)
|
||
+ trace_level_metrics_count,
|
||
)
|
||
|
||
start_time = time.perf_counter()
|
||
|
||
# On errored traces, skip trace-level metrics (no test case
|
||
# to judge) but DO run the span-level DFS walker below —
|
||
# it's what hydrates ``trace_api.*_spans`` for the dashboard,
|
||
# and per-span metric skip is handled inside ``dfs``.
|
||
skip_metrics_for_this_golden = False
|
||
if _skip_metrics_for_error(trace=current_trace):
|
||
trace_api.status = TraceSpanApiStatus.ERRORED
|
||
if progress and pbar_eval_id is not None:
|
||
update_pbar(
|
||
progress,
|
||
pbar_eval_id,
|
||
advance=count_total_metrics_for_trace(
|
||
current_trace
|
||
),
|
||
)
|
||
elif current_trace.metrics:
|
||
requires_trace = any(
|
||
metric.requires_trace
|
||
for metric in current_trace.metrics
|
||
)
|
||
|
||
# Build the trace-level LLMTestCase from the golden
|
||
# directly, the same way the async iterator does
|
||
# (see ``_a_evaluate_trace``). This makes top-level
|
||
# ``metrics=[...]`` work out of the box even when the
|
||
# user never calls ``update_current_trace(input=...)``.
|
||
llm_test_case = LLMTestCase(
|
||
input=golden.input,
|
||
actual_output=(
|
||
str(current_trace.output)
|
||
if current_trace.output is not None
|
||
else golden.actual_output
|
||
),
|
||
expected_output=current_trace.expected_output,
|
||
context=current_trace.context,
|
||
retrieval_context=current_trace.retrieval_context,
|
||
tools_called=current_trace.tools_called,
|
||
expected_tools=current_trace.expected_tools,
|
||
)
|
||
|
||
if requires_trace:
|
||
llm_test_case._trace_dict = (
|
||
trace_manager.create_nested_spans_dict(
|
||
current_trace.root_spans[0]
|
||
)
|
||
)
|
||
|
||
if not skip_metrics_for_this_golden:
|
||
for metric in current_trace.metrics:
|
||
metric.skipped = False
|
||
metric.error = None
|
||
if display_config.verbose_mode is not None:
|
||
metric.verbose_mode = (
|
||
display_config.verbose_mode
|
||
)
|
||
|
||
trace_api.metrics_data = []
|
||
for metric in current_trace.metrics:
|
||
res = _execute_metric(
|
||
metric=metric,
|
||
test_case=llm_test_case,
|
||
show_metric_indicator=show_metric_indicator,
|
||
in_component=True,
|
||
error_config=error_config,
|
||
)
|
||
if res == "skip":
|
||
continue
|
||
|
||
if not metric.skipped:
|
||
metric_data = create_metric_data(metric)
|
||
trace_api.metrics_data.append(metric_data)
|
||
api_test_case.update_metric_data(metric_data)
|
||
api_test_case.update_status(metric_data.success)
|
||
update_pbar(progress, pbar_eval_id)
|
||
|
||
# Always walk spans, even on errored traces — the walker
|
||
# hydrates ``trace_api.*_spans`` and the user needs that
|
||
# data on the dashboard to diagnose. Walk EVERY root, not
|
||
# just ``root_spans[0]``: OTel integrations can land
|
||
# multiple logical roots when a child ends before its
|
||
# parent. Mirrors the async path in ``agentic.py``.
|
||
for root in current_trace.root_spans:
|
||
dfs(root, progress, pbar_eval_id)
|
||
|
||
end_time = time.perf_counter()
|
||
run_duration = end_time - start_time
|
||
# Update test run
|
||
api_test_case.update_run_duration(run_duration)
|
||
test_run_manager.update_test_run(api_test_case, test_case)
|
||
main_result = create_test_result(api_test_case)
|
||
trace_results = extract_trace_test_results(trace_api)
|
||
unique_trace_results = filter_duplicate_results(
|
||
main_result, trace_results
|
||
)
|
||
test_results.append(main_result)
|
||
test_results.extend(unique_trace_results)
|
||
|
||
update_pbar(progress, pbar_id)
|
||
|
||
# Post-iteration guard: refuse a run that ran with no metric source
|
||
# at any level. Must happen AFTER the for-loop since span-level
|
||
# @observe metrics only become visible after user code has run.
|
||
if not _has_any_evaluable_metrics(
|
||
trace_metrics=trace_metrics,
|
||
traces=processed_traces,
|
||
test_case_metrics=trace_manager.eval_session.test_case_metrics,
|
||
):
|
||
_raise_no_metrics_error()
|
||
|
||
try:
|
||
if display_config.show_indicator and _use_bar_indicator:
|
||
progress = Progress(
|
||
TextColumn("{task.description}"),
|
||
BarColumn(bar_width=60),
|
||
TaskProgressColumn(),
|
||
TimeElapsedColumn(),
|
||
console=custom_console,
|
||
)
|
||
with progress:
|
||
pbar_id = add_pbar(
|
||
progress,
|
||
"Running Component-Level Evals (sync)",
|
||
total=len(goldens) * 2,
|
||
)
|
||
yield from evaluate_test_cases(
|
||
progress=progress, pbar_id=pbar_id
|
||
)
|
||
else:
|
||
yield from evaluate_test_cases()
|
||
except Exception:
|
||
raise
|
||
finally:
|
||
# Atomic exit cleanup: replacing the session resets mode + every
|
||
# per-run collection in a single assignment, so state can't leak
|
||
# into the next run.
|
||
local_trace_manager.eval_session = EvalSession()
|
||
|
||
|
||
def a_execute_agentic_test_cases_from_loop(
|
||
goldens: List[Golden],
|
||
trace_metrics: Optional[List[BaseMetric]],
|
||
test_results: List[TestResult],
|
||
loop: asyncio.AbstractEventLoop,
|
||
display_config: Optional[DisplayConfig] = DisplayConfig(),
|
||
cache_config: Optional[CacheConfig] = CacheConfig(),
|
||
error_config: Optional[ErrorConfig] = ErrorConfig(),
|
||
async_config: Optional[AsyncConfig] = AsyncConfig(),
|
||
identifier: Optional[str] = None,
|
||
_use_bar_indicator: bool = True,
|
||
_is_assert_test: bool = False,
|
||
) -> Iterator[TestResult]:
|
||
|
||
semaphore = asyncio.Semaphore(async_config.max_concurrent)
|
||
original_create_task = asyncio.create_task
|
||
|
||
test_run_manager = global_test_run_manager
|
||
test_run_manager.save_to_disk = cache_config.write_cache
|
||
test_run = test_run_manager.get_test_run(identifier=identifier)
|
||
|
||
local_trace_manager = trace_manager
|
||
local_trace_manager.eval_session = EvalSession(mode=EvalMode.ITERATOR_ASYNC)
|
||
|
||
async def execute_callback_with_semaphore(coroutine: Awaitable):
|
||
async with semaphore:
|
||
return await _await_with_outer_deadline(
|
||
coroutine, timeout=get_per_task_timeout_seconds()
|
||
)
|
||
|
||
def evaluate_test_cases(
|
||
progress: Optional[Progress] = None,
|
||
pbar_id: Optional[int] = None,
|
||
pbar_callback_id: Optional[int] = None,
|
||
):
|
||
# Tasks we scheduled during this iterator run on this event loop.
|
||
# by gathering these tasks we can avoid re-awaiting coroutines which
|
||
# can cause cross loop mixups that trigger "future belongs to a different loop" errors
|
||
created_tasks: list[asyncio.Task] = []
|
||
task_meta: dict[asyncio.Task, dict] = {}
|
||
current_golden_ctx = {"index": -1, "name": None, "input": None}
|
||
|
||
def create_callback_task(coro, **kwargs):
|
||
# build a descriptive task name for tracking
|
||
coro_desc = repr(coro)
|
||
task_name = f"callback[{current_golden_ctx['index']}]:{coro_desc.split()[1] if ' ' in coro_desc else coro_desc}"
|
||
|
||
# Wrap the user coroutine in our semaphore runner and bind it to THIS loop.
|
||
# Keep the resulting Task so we can gather tasks (not raw coroutines) later,
|
||
# without touching tasks from other loops or already awaited coroutines.
|
||
task = loop.create_task(
|
||
execute_callback_with_semaphore(coro), name=task_name
|
||
)
|
||
|
||
# record metadata for debugging
|
||
started = time.perf_counter()
|
||
short_input = current_golden_ctx.get("input")
|
||
if isinstance(short_input, str):
|
||
short_input = shorten(short_input, len_medium())
|
||
|
||
task_meta[task] = {
|
||
"golden_index": current_golden_ctx["index"],
|
||
"golden_name": current_golden_ctx["name"],
|
||
"input": short_input,
|
||
"coro": coro_desc,
|
||
"started": started,
|
||
}
|
||
|
||
def on_task_done(t: asyncio.Task):
|
||
cancelled = False
|
||
exc = None
|
||
trace = None
|
||
root = None
|
||
resolved_trace_from_task = False
|
||
resolved_root_from_task = False
|
||
|
||
# Task.exception() raises CancelledError if task was cancelled
|
||
try:
|
||
exc = t.exception()
|
||
except asyncio.CancelledError:
|
||
cancelled = True
|
||
exc = None
|
||
|
||
meta = task_meta.get(t, {})
|
||
golden_index = meta.get("golden_index")
|
||
|
||
if golden_index is not None and 0 <= golden_index < len(
|
||
goldens
|
||
):
|
||
golden = goldens[golden_index]
|
||
|
||
def _mark_trace_error(trace, root, msg: str):
|
||
now = time.perf_counter()
|
||
trace.status = TraceSpanStatus.ERRORED
|
||
# Close the trace so the API layer has a proper endTime
|
||
if trace.end_time is None:
|
||
trace.end_time = now
|
||
if root:
|
||
root.status = TraceSpanStatus.ERRORED
|
||
root.error = msg
|
||
if root.end_time is None:
|
||
root.end_time = now
|
||
|
||
if exc is not None:
|
||
msg = format_error_text(exc)
|
||
trace, root = _resolve_trace_and_root_for_task(t)
|
||
resolved_trace_from_task = bool(trace)
|
||
resolved_root_from_task = bool(root)
|
||
if trace:
|
||
_mark_trace_error(trace, root, msg)
|
||
else:
|
||
for (
|
||
trace
|
||
) in trace_manager.eval_session.traces_to_evaluate:
|
||
if (
|
||
trace_manager.eval_session.trace_uuid_to_golden.get(
|
||
trace.uuid
|
||
)
|
||
is golden
|
||
):
|
||
root = _pick_root_for_marking(trace)
|
||
_mark_trace_error(trace, root, msg)
|
||
break
|
||
|
||
elif cancelled or t.cancelled():
|
||
cancel_exc = DeepEvalError(
|
||
"Task was cancelled (likely due to timeout)."
|
||
)
|
||
msg = format_error_text(cancel_exc)
|
||
trace, root = _resolve_trace_and_root_for_task(t)
|
||
resolved_trace_from_task = bool(trace)
|
||
resolved_root_from_task = bool(root)
|
||
if trace:
|
||
_mark_trace_error(trace, root, msg)
|
||
else:
|
||
for (
|
||
trace
|
||
) in trace_manager.eval_session.traces_to_evaluate:
|
||
if (
|
||
trace_manager.eval_session.trace_uuid_to_golden.get(
|
||
trace.uuid
|
||
)
|
||
is golden
|
||
):
|
||
root = _pick_root_for_marking(trace)
|
||
_mark_trace_error(trace, root, msg)
|
||
break
|
||
|
||
if get_settings().DEEPEVAL_DEBUG_ASYNC:
|
||
# Using info level here to make it easy to spot these logs.
|
||
golden_name = meta.get("golden_name")
|
||
duration = time.perf_counter() - meta.get(
|
||
"started", started
|
||
)
|
||
|
||
if cancelled or exc is not None:
|
||
if not resolved_trace_from_task:
|
||
logger.warning(
|
||
"[deepeval] on_task_done: no binding for task; falling back to golden->trace. task=%s golden=%r",
|
||
t.get_name(),
|
||
golden_name,
|
||
)
|
||
elif not resolved_root_from_task:
|
||
logger.warning(
|
||
"[deepeval] on_task_done: bound trace found but no bound root; using heuristic. task=%s trace=%s",
|
||
t.get_name(),
|
||
trace.uuid,
|
||
)
|
||
|
||
if cancelled:
|
||
logger.info(
|
||
"[deepeval] task CANCELLED %s after %.2fs meta=%r",
|
||
t.get_name(),
|
||
duration,
|
||
meta,
|
||
)
|
||
elif exc is not None:
|
||
|
||
show_trace = bool(
|
||
get_settings().DEEPEVAL_LOG_STACK_TRACES
|
||
)
|
||
exc_info = (
|
||
(
|
||
type(exc),
|
||
exc,
|
||
getattr(exc, "__traceback__", None),
|
||
)
|
||
if show_trace
|
||
else None
|
||
)
|
||
logger.error(
|
||
"[deepeval] task ERROR %s after %.2fs meta=%r",
|
||
t.get_name(),
|
||
duration,
|
||
meta,
|
||
exc_info=exc_info,
|
||
)
|
||
else:
|
||
logger.info(
|
||
"[deepeval] task OK %s after %.2fs meta={'golden_index': %r}",
|
||
t.get_name(),
|
||
duration,
|
||
meta.get("golden_index"),
|
||
)
|
||
|
||
try:
|
||
trace_manager.task_bindings.pop(t, None)
|
||
except Exception:
|
||
pass
|
||
update_pbar(progress, pbar_callback_id)
|
||
update_pbar(progress, pbar_id)
|
||
|
||
task.add_done_callback(on_task_done)
|
||
created_tasks.append(task)
|
||
return task
|
||
|
||
asyncio.create_task = create_callback_task
|
||
# DEBUG
|
||
# Snapshot tasks that already exist on this loop so we can detect strays
|
||
baseline_tasks = loop.run_until_complete(_snapshot_tasks())
|
||
|
||
try:
|
||
for index, golden in enumerate(goldens):
|
||
token = set_current_golden(golden)
|
||
current_golden_ctx.update(
|
||
{
|
||
"index": index,
|
||
"name": getattr(golden, "name", None),
|
||
"input": getattr(golden, "input", None),
|
||
}
|
||
)
|
||
prev_task_length = len(created_tasks)
|
||
try:
|
||
yield golden
|
||
finally:
|
||
reset_current_golden(token)
|
||
# if this golden created no tasks, bump bars now
|
||
if len(created_tasks) == prev_task_length:
|
||
update_pbar(progress, pbar_callback_id)
|
||
update_pbar(progress, pbar_id)
|
||
finally:
|
||
asyncio.create_task = original_create_task
|
||
|
||
if created_tasks:
|
||
# Only await tasks we created on this loop in this run.
|
||
# This will prevent re-awaiting and avoids cross loop "future belongs to a different loop" errors
|
||
try:
|
||
loop.run_until_complete(
|
||
asyncio.wait_for(
|
||
asyncio.gather(*created_tasks, return_exceptions=True),
|
||
timeout=get_gather_timeout(),
|
||
)
|
||
)
|
||
|
||
except (asyncio.TimeoutError, TimeoutError) as e:
|
||
import traceback
|
||
|
||
settings = get_settings()
|
||
pending = [t for t in created_tasks if not t.done()]
|
||
|
||
_log_gather_timeout(logger, exc=e, pending=len(pending))
|
||
|
||
# Log the elapsed time for each task that was pending
|
||
for t in pending:
|
||
meta = task_meta.get(t, {})
|
||
start_time = meta.get("started", time.perf_counter())
|
||
elapsed_time = time.perf_counter() - start_time
|
||
|
||
# Determine if it was a per task or gather timeout based on task's elapsed time
|
||
if not settings.DEEPEVAL_DISABLE_TIMEOUTS:
|
||
timeout_type = (
|
||
"per-task"
|
||
if elapsed_time >= get_per_task_timeout_seconds()
|
||
else "gather"
|
||
)
|
||
logger.info(
|
||
" - PENDING %s elapsed_time=%.2fs timeout_type=%s meta=%s",
|
||
t.get_name(),
|
||
elapsed_time,
|
||
timeout_type,
|
||
meta,
|
||
)
|
||
else:
|
||
logger.info(
|
||
" - PENDING %s elapsed_time=%.2fs meta=%s",
|
||
t.get_name(),
|
||
elapsed_time,
|
||
meta,
|
||
)
|
||
|
||
if loop.get_debug() and get_settings().DEEPEVAL_DEBUG_ASYNC:
|
||
frames = t.get_stack(limit=6)
|
||
if frames:
|
||
logger.info(" stack:")
|
||
for fr in frames:
|
||
for line in traceback.format_stack(fr):
|
||
logger.info(" " + line.rstrip())
|
||
|
||
# Cancel and drain the tasks
|
||
for t in pending:
|
||
t.cancel()
|
||
loop.run_until_complete(
|
||
asyncio.gather(*created_tasks, return_exceptions=True)
|
||
)
|
||
finally:
|
||
|
||
# if it is already closed, we are done
|
||
if loop.is_closed():
|
||
return
|
||
|
||
try:
|
||
current_tasks = set()
|
||
# Find tasks that were created during this run but we didn’t track
|
||
current_tasks = loop.run_until_complete(_snapshot_tasks())
|
||
except RuntimeError:
|
||
# this might happen if the loop is already closing
|
||
pass
|
||
|
||
leftovers = [
|
||
t
|
||
for t in current_tasks
|
||
if t not in baseline_tasks
|
||
and t not in created_tasks
|
||
and not t.done()
|
||
]
|
||
|
||
if get_settings().DEEPEVAL_DEBUG_ASYNC:
|
||
if len(leftovers) > 0:
|
||
logger.warning(
|
||
"[deepeval] %d stray task(s) not tracked; cancelling...",
|
||
len(leftovers),
|
||
)
|
||
for t in leftovers:
|
||
meta = task_meta.get(t, {})
|
||
name = t.get_name()
|
||
logger.warning(" - STRAY %s meta=%s", name, meta)
|
||
|
||
if leftovers:
|
||
for t in leftovers:
|
||
t.cancel()
|
||
|
||
# Drain strays so they don’t leak into the next iteration
|
||
try:
|
||
loop.run_until_complete(
|
||
asyncio.gather(*leftovers, return_exceptions=True)
|
||
)
|
||
except RuntimeError:
|
||
# If the loop is closing here, just continue
|
||
if get_settings().DEEPEVAL_DEBUG_ASYNC:
|
||
logger.warning(
|
||
"[deepeval] failed to drain stray tasks because loop is closing"
|
||
)
|
||
|
||
# Pre-evaluation guard: refuse a run that has no metric source.
|
||
# Lazy check is the only correct option because span-level metrics
|
||
# on @observe-decorated functions only become visible after user
|
||
# code has actually run.
|
||
session = trace_manager.eval_session
|
||
if not _has_any_evaluable_metrics(
|
||
trace_metrics=trace_metrics,
|
||
traces=session.traces_to_evaluate,
|
||
test_case_metrics=session.test_case_metrics,
|
||
):
|
||
_raise_no_metrics_error()
|
||
|
||
# Evaluate traces
|
||
if trace_manager.eval_session.traces_to_evaluate:
|
||
loop.run_until_complete(
|
||
_a_evaluate_traces(
|
||
traces_to_evaluate=trace_manager.eval_session.traces_to_evaluate,
|
||
goldens=goldens,
|
||
test_run_manager=test_run_manager,
|
||
test_results=test_results,
|
||
trace_metrics=trace_metrics,
|
||
verbose_mode=display_config.verbose_mode,
|
||
ignore_errors=error_config.ignore_errors,
|
||
skip_on_missing_params=error_config.skip_on_missing_params,
|
||
show_indicator=display_config.show_indicator,
|
||
throttle_value=async_config.throttle_value,
|
||
max_concurrent=async_config.max_concurrent,
|
||
_use_bar_indicator=_use_bar_indicator,
|
||
_is_assert_test=_is_assert_test,
|
||
progress=progress,
|
||
pbar_id=pbar_id,
|
||
)
|
||
)
|
||
elif trace_manager.eval_session.test_case_metrics:
|
||
loop.run_until_complete(
|
||
_evaluate_test_case_pairs(
|
||
test_case_pairs=trace_manager.eval_session.test_case_metrics,
|
||
test_run=test_run,
|
||
test_run_manager=test_run_manager,
|
||
test_results=test_results,
|
||
ignore_errors=error_config.ignore_errors,
|
||
skip_on_missing_params=error_config.skip_on_missing_params,
|
||
show_indicator=display_config.show_indicator,
|
||
verbose_mode=display_config.verbose_mode,
|
||
throttle_value=async_config.throttle_value,
|
||
max_concurrent=async_config.max_concurrent,
|
||
_use_bar_indicator=_use_bar_indicator,
|
||
_is_assert_test=_is_assert_test,
|
||
progress=progress,
|
||
pbar_id=pbar_id,
|
||
)
|
||
)
|
||
|
||
try:
|
||
if display_config.show_indicator and _use_bar_indicator:
|
||
progress = Progress(
|
||
TextColumn("{task.description}"),
|
||
BarColumn(bar_width=60),
|
||
TaskProgressColumn(),
|
||
TimeElapsedColumn(),
|
||
console=custom_console,
|
||
)
|
||
with progress:
|
||
pbar_id = add_pbar(
|
||
progress,
|
||
"Running Component-Level Evals (async)",
|
||
total=len(goldens) * 2,
|
||
)
|
||
pbar_callback_id = add_pbar(
|
||
progress,
|
||
f"\t⚡ Calling LLM app (with {len(goldens)} goldens)",
|
||
total=len(goldens),
|
||
)
|
||
yield from evaluate_test_cases(
|
||
progress=progress,
|
||
pbar_id=pbar_id,
|
||
pbar_callback_id=pbar_callback_id,
|
||
)
|
||
else:
|
||
yield from evaluate_test_cases()
|
||
except Exception:
|
||
raise
|
||
finally:
|
||
# Atomic exit cleanup: replacing the session resets mode + every
|
||
# per-run collection in a single assignment.
|
||
local_trace_manager.eval_session = EvalSession()
|
||
|
||
|
||
async def _a_evaluate_traces(
|
||
traces_to_evaluate: List[Trace],
|
||
goldens: List[Golden],
|
||
test_run_manager: TestRunManager,
|
||
test_results: List[TestResult],
|
||
verbose_mode: Optional[bool],
|
||
ignore_errors: bool,
|
||
skip_on_missing_params: bool,
|
||
show_indicator: bool,
|
||
_use_bar_indicator: bool,
|
||
_is_assert_test: bool,
|
||
progress: Optional[Progress],
|
||
pbar_id: Optional[int],
|
||
throttle_value: int,
|
||
max_concurrent: int,
|
||
trace_metrics: Optional[List[BaseMetric]],
|
||
):
|
||
semaphore = asyncio.Semaphore(max_concurrent)
|
||
|
||
async def execute_evals_with_semaphore(func: Callable, *args, **kwargs):
|
||
async with semaphore:
|
||
return await _await_with_outer_deadline(
|
||
func, *args, timeout=get_per_task_timeout_seconds(), **kwargs
|
||
)
|
||
|
||
eval_tasks = []
|
||
# Here, we will work off a fixed-set copy to avoid surprises from potential
|
||
# mid-iteration mutation
|
||
traces_snapshot = list(traces_to_evaluate or [])
|
||
|
||
for count, trace in enumerate(traces_snapshot):
|
||
# Prefer the explicit mapping from trace -> golden captured at trace creation.
|
||
golden = trace_manager.eval_session.trace_uuid_to_golden.get(trace.uuid)
|
||
if not golden:
|
||
# trace started during the iterator run but the CURRENT_GOLDEN was
|
||
# not set for some reason. We can’t map it to a golden, so the best
|
||
# we can do is skip evaluation for this trace.
|
||
if (
|
||
logger.isEnabledFor(logging.DEBUG)
|
||
and get_settings().DEEPEVAL_VERBOSE_MODE
|
||
):
|
||
logger.debug(
|
||
"Skipping trace %s: no golden association found in eval_session",
|
||
trace.uuid,
|
||
)
|
||
continue
|
||
copied_trace_metrics: Optional[List[BaseMetric]] = None
|
||
if trace_metrics:
|
||
copied_trace_metrics = copy_metrics(trace_metrics)
|
||
with capture_evaluation_run("golden"):
|
||
task = execute_evals_with_semaphore(
|
||
func=_a_execute_agentic_test_case,
|
||
golden=golden,
|
||
trace=trace,
|
||
test_run_manager=test_run_manager,
|
||
test_results=test_results,
|
||
count=count,
|
||
verbose_mode=verbose_mode,
|
||
ignore_errors=ignore_errors,
|
||
skip_on_missing_params=skip_on_missing_params,
|
||
show_indicator=show_indicator,
|
||
_use_bar_indicator=_use_bar_indicator,
|
||
_is_assert_test=_is_assert_test,
|
||
progress=progress,
|
||
pbar_id=pbar_id,
|
||
trace_metrics=copied_trace_metrics,
|
||
)
|
||
eval_tasks.append(asyncio.create_task(task))
|
||
await asyncio.sleep(throttle_value)
|
||
|
||
try:
|
||
await asyncio.wait_for(
|
||
asyncio.gather(*eval_tasks),
|
||
timeout=get_gather_timeout(),
|
||
)
|
||
except (asyncio.TimeoutError, TimeoutError):
|
||
for t in eval_tasks:
|
||
if not t.done():
|
||
t.cancel()
|
||
await asyncio.gather(*eval_tasks, return_exceptions=True)
|
||
raise
|