Executor

Executor(desc: str = 'Evaluating', show_progress: bool = True, keep_progress_bar: bool = True, jobs: List[Any] = list(), raise_exceptions: bool = False, batch_size: Optional[int] = None, run_config: Optional[RunConfig] = None, pbar: Optional[tqdm] = None, _jobs_processed: int = 0, _cancel_event: Event = Event())

用于运行异步任务的执行器类，支持进度跟踪与错误处理。

属性：

cancel

cancel() -> None

取消所有任务的执行。

源代码位于 src/ragas/executor.py

def cancel(self) -> None:
    """Cancel the execution of all jobs."""
    self._cancel_event.set()

is_cancelled

is_cancelled() -> bool

检查执行是否已被取消。

源代码位于 src/ragas/executor.py

def is_cancelled(self) -> bool:
    """Check if the execution has been cancelled."""
    return self._cancel_event.is_set()

submit

submit(callable: Callable, *args, name: Optional[str] = None, **kwargs) -> None

提交一个待执行的任务，对可调用对象进行错误处理包装，并编号以便跟踪任务索引。

源代码位于 src/ragas/executor.py

def submit(
    self,
    callable: t.Callable,
    *args,
    name: t.Optional[str] = None,
    **kwargs,
) -> None:
    """
    Submit a job to be executed, wrapping the callable with error handling and indexing to keep track of the job index.
    """
    # Use _jobs_processed for consistent indexing across multiple runs
    callable_with_index = self.wrap_callable_with_index(
        callable, self._jobs_processed
    )
    self.jobs.append((callable_with_index, args, kwargs, name))
    self._jobs_processed += 1

clear_jobs

clear_jobs() -> None

清除所有已提交任务并重置计数器。

源代码位于 src/ragas/executor.py

def clear_jobs(self) -> None:
    """Clear all submitted jobs and reset counter."""
    self.jobs.clear()
    self._jobs_processed = 0

aresults

aresults() -> List[Any]

异步执行所有已提交任务并返回结果。结果按任务提交顺序返回。

这是已处于异步上下文中时执行异步任务的异步入口。

源代码位于 src/ragas/executor.py

async def aresults(self) -> t.List[t.Any]:
    """
    Execute all submitted jobs and return their results asynchronously.
    The results are returned in the order of job submission.

    This is the async entry point for executing async jobs when already in an async context.
    """
    results = await self._process_jobs()
    sorted_results = sorted(results, key=lambda x: x[0])
    return [r[1] for r in sorted_results]

results

results() -> List[Any]

执行所有已提交任务并返回结果。结果按任务提交顺序返回。

这是执行异步任务的主要同步入口。

源代码位于 src/ragas/executor.py

def results(self) -> t.List[t.Any]:
    """
    Execute all submitted jobs and return their results. The results are returned in the order of job submission.

    This is the main sync entry point for executing async jobs.
    """

    async def _async_wrapper():
        return await self.aresults()

    apply_nest_asyncio()
    return run(_async_wrapper)

run_async_batch

run_async_batch(desc: str, func: Callable, kwargs_list: List[Dict], batch_size: Optional[int] = None)

提供以不同参数并行运行同一异步函数的能力。

源代码位于 src/ragas/executor.py

def run_async_batch(
    desc: str,
    func: t.Callable,
    kwargs_list: t.List[t.Dict],
    batch_size: t.Optional[int] = None,
):
    """
    Provide functionality to run the same async function with different arguments in parallel.
    """
    run_config = RunConfig()
    executor = Executor(
        desc=desc,
        keep_progress_bar=False,
        raise_exceptions=True,
        run_config=run_config,
        batch_size=batch_size,
    )

    for kwargs in kwargs_list:
        executor.submit(func, **kwargs)

    return executor.results()