索引

评估模块。

BaseEvaluator

基类: PromptMixin

基础评估器类。

源代码位于 llama-index-core/llama_index/core/evaluation/base.py

class BaseEvaluator(PromptMixin):
    """Base Evaluator class."""

    def _get_prompt_modules(self) -> PromptMixinType:
        """Get prompt modules."""
        return {}

    def evaluate(
        self,
        query: Optional[str] = None,
        response: Optional[str] = None,
        contexts: Optional[Sequence[str]] = None,
        **kwargs: Any,
    ) -> EvaluationResult:
        """
        Run evaluation with query string, retrieved contexts,
        and generated response string.

        Subclasses can override this method to provide custom evaluation logic and
        take in additional arguments.
        """
        return asyncio_run(
            self.aevaluate(
                query=query,
                response=response,
                contexts=contexts,
                **kwargs,
            )
        )

    @abstractmethod
    async def aevaluate(
        self,
        query: Optional[str] = None,
        response: Optional[str] = None,
        contexts: Optional[Sequence[str]] = None,
        **kwargs: Any,
    ) -> EvaluationResult:
        """
        Run evaluation with query string, retrieved contexts,
        and generated response string.

        Subclasses can override this method to provide custom evaluation logic and
        take in additional arguments.
        """
        raise NotImplementedError

    def evaluate_response(
        self,
        query: Optional[str] = None,
        response: Optional[Response] = None,
        **kwargs: Any,
    ) -> EvaluationResult:
        """
        Run evaluation with query string and generated Response object.

        Subclasses can override this method to provide custom evaluation logic and
        take in additional arguments.
        """
        response_str: Optional[str] = None
        contexts: Optional[Sequence[str]] = None
        if response is not None:
            response_str = response.response
            contexts = [node.get_content() for node in response.source_nodes]

        return self.evaluate(
            query=query, response=response_str, contexts=contexts, **kwargs
        )

    async def aevaluate_response(
        self,
        query: Optional[str] = None,
        response: Optional[Response] = None,
        **kwargs: Any,
    ) -> EvaluationResult:
        """
        Run evaluation with query string and generated Response object.

        Subclasses can override this method to provide custom evaluation logic and
        take in additional arguments.
        """
        response_str: Optional[str] = None
        contexts: Optional[Sequence[str]] = None
        if response is not None:
            response_str = response.response
            contexts = [node.get_content() for node in response.source_nodes]

        return await self.aevaluate(
            query=query, response=response_str, contexts=contexts, **kwargs
        )

evaluate

evaluate(query: Optional[str] = None, response: Optional[str] = None, contexts: Optional[Sequence[str]] = None, **kwargs: Any) -> EvaluationResult

使用查询字符串、检索到的上下文和生成的响应字符串运行评估。

子类可以重写此方法以提供自定义评估逻辑并接受额外的参数。

源代码位于 llama-index-core/llama_index/core/evaluation/base.py

def evaluate(
    self,
    query: Optional[str] = None,
    response: Optional[str] = None,
    contexts: Optional[Sequence[str]] = None,
    **kwargs: Any,
) -> EvaluationResult:
    """
    Run evaluation with query string, retrieved contexts,
    and generated response string.

    Subclasses can override this method to provide custom evaluation logic and
    take in additional arguments.
    """
    return asyncio_run(
        self.aevaluate(
            query=query,
            response=response,
            contexts=contexts,
            **kwargs,
        )
    )

aevaluate abstractmethod async

aevaluate(query: Optional[str] = None, response: Optional[str] = None, contexts: Optional[Sequence[str]] = None, **kwargs: Any) -> EvaluationResult

使用查询字符串、检索到的上下文和生成的响应字符串运行评估。

子类可以重写此方法以提供自定义评估逻辑并接受额外的参数。

源代码位于 llama-index-core/llama_index/core/evaluation/base.py

@abstractmethod
async def aevaluate(
    self,
    query: Optional[str] = None,
    response: Optional[str] = None,
    contexts: Optional[Sequence[str]] = None,
    **kwargs: Any,
) -> EvaluationResult:
    """
    Run evaluation with query string, retrieved contexts,
    and generated response string.

    Subclasses can override this method to provide custom evaluation logic and
    take in additional arguments.
    """
    raise NotImplementedError

evaluate_response

evaluate_response(query: Optional[str] = None, response: Optional[Response] = None, **kwargs: Any) -> EvaluationResult

使用查询字符串和生成的 Response 对象运行评估。

子类可以重写此方法以提供自定义评估逻辑并接受额外的参数。

源代码位于 llama-index-core/llama_index/core/evaluation/base.py

def evaluate_response(
    self,
    query: Optional[str] = None,
    response: Optional[Response] = None,
    **kwargs: Any,
) -> EvaluationResult:
    """
    Run evaluation with query string and generated Response object.

    Subclasses can override this method to provide custom evaluation logic and
    take in additional arguments.
    """
    response_str: Optional[str] = None
    contexts: Optional[Sequence[str]] = None
    if response is not None:
        response_str = response.response
        contexts = [node.get_content() for node in response.source_nodes]

    return self.evaluate(
        query=query, response=response_str, contexts=contexts, **kwargs
    )

aevaluate_response async

aevaluate_response(query: Optional[str] = None, response: Optional[Response] = None, **kwargs: Any) -> EvaluationResult

使用查询字符串和生成的 Response 对象运行评估。

子类可以重写此方法以提供自定义评估逻辑并接受额外的参数。

源代码位于 llama-index-core/llama_index/core/evaluation/base.py

async def aevaluate_response(
    self,
    query: Optional[str] = None,
    response: Optional[Response] = None,
    **kwargs: Any,
) -> EvaluationResult:
    """
    Run evaluation with query string and generated Response object.

    Subclasses can override this method to provide custom evaluation logic and
    take in additional arguments.
    """
    response_str: Optional[str] = None
    contexts: Optional[Sequence[str]] = None
    if response is not None:
        response_str = response.response
        contexts = [node.get_content() for node in response.source_nodes]

    return await self.aevaluate(
        query=query, response=response_str, contexts=contexts, **kwargs
    )

EvaluationResult

基类: BaseModel

评估结果。

BaseEvaluator 的输出。

参数

名称	类型	描述	默认值
query	str | None	查询字符串	无
contexts	Sequence[str] | None	上下文字符串	无
response	str | None	响应字符串	无
passing	bool | None	二元评估结果（通过或未通过）	无
feedback	str | None	对响应的反馈或理由	无
score	float | None	响应得分	无
pairwise_source	str | None	仅用于成对评估，指定结果是来自提供答案的原始顺序还是翻转顺序	无
invalid_result	bool	评估结果是否无效。	False
invalid_reason	str | None	评估无效的原因。	无

源代码位于 llama-index-core/llama_index/core/evaluation/base.py

class EvaluationResult(BaseModel):
    """
    Evaluation result.

    Output of an BaseEvaluator.
    """

    query: Optional[str] = Field(default=None, description="Query string")
    contexts: Optional[Sequence[str]] = Field(
        default=None, description="Context strings"
    )
    response: Optional[str] = Field(default=None, description="Response string")
    passing: Optional[bool] = Field(
        default=None, description="Binary evaluation result (passing or not)"
    )
    feedback: Optional[str] = Field(
        default=None, description="Feedback or reasoning for the response"
    )
    score: Optional[float] = Field(default=None, description="Score for the response")
    pairwise_source: Optional[str] = Field(
        default=None,
        description=(
            "Used only for pairwise and specifies whether it is from original order of"
            " presented answers or flipped order"
        ),
    )
    invalid_result: bool = Field(
        default=False, description="Whether the evaluation result is an invalid one."
    )
    invalid_reason: Optional[str] = Field(
        default=None, description="Reason for invalid evaluation."
    )

BatchEvalRunner

批量评估运行器。

参数

名称	类型	描述	默认值
evaluators	Dict[str, BaseEvaluator]	评估器字典。	required
workers	int	用于并行化的工作线程数。默认为 2。	2
show_progress	bool	是否显示进度条。默认为 False。	False

源代码位于 llama-index-core/llama_index/core/evaluation/batch_runner.py

class BatchEvalRunner:
    """
    Batch evaluation runner.

    Args:
        evaluators (Dict[str, BaseEvaluator]): Dictionary of evaluators.
        workers (int): Number of workers to use for parallelization.
            Defaults to 2.
        show_progress (bool): Whether to show progress bars. Defaults to False.

    """

    def __init__(
        self,
        evaluators: Dict[str, BaseEvaluator],
        workers: int = 2,
        show_progress: bool = False,
    ):
        self.evaluators = evaluators
        self.workers = workers
        self.semaphore = asyncio.Semaphore(self.workers)
        self.show_progress = show_progress
        self.asyncio_mod = asyncio_module(show_progress=self.show_progress)

    def _format_results(
        self, results: List[Tuple[str, EvaluationResult]]
    ) -> Dict[str, List[EvaluationResult]]:
        """Format results."""
        # Format results
        results_dict: Dict[str, List[EvaluationResult]] = {
            name: [] for name in self.evaluators
        }
        for name, result in results:
            results_dict[name].append(result)

        return results_dict

    def _validate_and_clean_inputs(
        self,
        *inputs_list: Any,
    ) -> List[Any]:
        """
        Validate and clean input lists.

        Enforce that at least one of the inputs is not None.
        Make sure that all inputs have the same length.
        Make sure that None inputs are replaced with [None] * len(inputs).

        """
        assert len(inputs_list) > 0
        # first, make sure at least one of queries or response_strs is not None
        input_len: Optional[int] = None
        for inputs in inputs_list:
            if inputs is not None:
                input_len = len(inputs)
                break
        if input_len is None:
            raise ValueError("At least one item in inputs_list must be provided.")

        new_inputs_list = []
        for inputs in inputs_list:
            if inputs is None:
                new_inputs_list.append([None] * input_len)
            else:
                if len(inputs) != input_len:
                    raise ValueError("All inputs must have the same length.")
                new_inputs_list.append(inputs)
        return new_inputs_list

    def _validate_nested_eval_kwargs_types(
        self, eval_kwargs_lists: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        Ensure eval kwargs are acceptable format.
            either a Dict[str, List] or a Dict[str, Dict[str, List]].

        Allows use of different kwargs (e.g. references) with different evaluators
            while keeping backwards compatibility for single evaluators

        """
        if not isinstance(eval_kwargs_lists, dict):
            raise ValueError(
                f"eval_kwargs_lists must be a dict. Got {eval_kwargs_lists}"
            )

        for evaluator, eval_kwargs in eval_kwargs_lists.items():
            if isinstance(eval_kwargs, list):
                # maintain backwards compatibility - for use with single evaluator
                eval_kwargs_lists[evaluator] = self._validate_and_clean_inputs(
                    eval_kwargs
                )[0]
            elif isinstance(eval_kwargs, dict):
                # for use with multiple evaluators
                for k in eval_kwargs:
                    v = eval_kwargs[k]
                    if not isinstance(v, list):
                        raise ValueError(
                            f"nested inner values in eval_kwargs must be a list. Got {evaluator}: {k}: {v}"
                        )
                    eval_kwargs_lists[evaluator][k] = self._validate_and_clean_inputs(
                        v
                    )[0]
            else:
                raise ValueError(
                    f"eval_kwargs must be a list or a dict. Got {evaluator}: {eval_kwargs}"
                )
        return eval_kwargs_lists

    def _get_eval_kwargs(
        self, eval_kwargs_lists: Dict[str, Any], idx: int
    ) -> Dict[str, Any]:
        """
        Get eval kwargs from eval_kwargs_lists at a given idx.

        Since eval_kwargs_lists is a dict of lists, we need to get the
        value at idx for each key.

        """
        return {k: v[idx] for k, v in eval_kwargs_lists.items()}

    async def aevaluate_response_strs(
        self,
        queries: Optional[List[str]] = None,
        response_strs: Optional[List[str]] = None,
        contexts_list: Optional[List[List[str]]] = None,
        **eval_kwargs_lists: Dict[str, Any],
    ) -> Dict[str, List[EvaluationResult]]:
        """
        Evaluate query, response pairs.

        This evaluates queries, responses, contexts as string inputs.
        Can supply additional kwargs to the evaluator in eval_kwargs_lists.

        Args:
            queries (Optional[List[str]]): List of query strings. Defaults to None.
            response_strs (Optional[List[str]]): List of response strings.
                Defaults to None.
            contexts_list (Optional[List[List[str]]]): List of context lists.
                Defaults to None.
            **eval_kwargs_lists (Dict[str, Any]): Dict of either dicts or lists
                of kwargs to pass to evaluator. Defaults to None.
                    multiple evaluators: {evaluator: {kwarg: [list of values]},...}
                    single evaluator:    {kwarg: [list of values]}

        """
        queries, response_strs, contexts_list = self._validate_and_clean_inputs(
            queries, response_strs, contexts_list
        )
        eval_kwargs_lists = self._validate_nested_eval_kwargs_types(eval_kwargs_lists)

        # boolean to check if using multi kwarg evaluator
        multi_kwargs = len(eval_kwargs_lists) > 0 and isinstance(
            next(iter(eval_kwargs_lists.values())), dict
        )

        # run evaluations
        eval_jobs = []
        for idx, query in enumerate(cast(List[str], queries)):
            response_str = cast(List, response_strs)[idx]
            contexts = cast(List, contexts_list)[idx]
            for name, evaluator in self.evaluators.items():
                if multi_kwargs:
                    # multi-evaluator - get appropriate runtime kwargs if present
                    kwargs = (
                        eval_kwargs_lists[name] if name in eval_kwargs_lists else {}
                    )
                else:
                    # single evaluator (maintain backwards compatibility)
                    kwargs = eval_kwargs_lists
                eval_kwargs = self._get_eval_kwargs(kwargs, idx)
                eval_jobs.append(
                    eval_worker(
                        self.semaphore,
                        evaluator,
                        name,
                        query=query,
                        response_str=response_str,
                        contexts=contexts,
                        eval_kwargs=eval_kwargs,
                    )
                )
        results = await self.asyncio_mod.gather(*eval_jobs)

        # Format results
        return self._format_results(results)

    async def aevaluate_responses(
        self,
        queries: Optional[List[str]] = None,
        responses: Optional[List[Response]] = None,
        **eval_kwargs_lists: Dict[str, Any],
    ) -> Dict[str, List[EvaluationResult]]:
        """
        Evaluate query, response pairs.

        This evaluates queries and response objects.

        Args:
            queries (Optional[List[str]]): List of query strings. Defaults to None.
            responses (Optional[List[Response]]): List of response objects.
                Defaults to None.
            **eval_kwargs_lists (Dict[str, Any]): Dict of either dicts or lists
                of kwargs to pass to evaluator. Defaults to None.
                    multiple evaluators: {evaluator: {kwarg: [list of values]},...}
                    single evaluator:    {kwarg: [list of values]}

        """
        queries, responses = self._validate_and_clean_inputs(queries, responses)
        eval_kwargs_lists = self._validate_nested_eval_kwargs_types(eval_kwargs_lists)

        # boolean to check if using multi kwarg evaluator
        multi_kwargs = len(eval_kwargs_lists) > 0 and isinstance(
            next(iter(eval_kwargs_lists.values())), dict
        )

        # run evaluations
        eval_jobs = []
        for idx, query in enumerate(cast(List[str], queries)):
            response = cast(List, responses)[idx]
            for name, evaluator in self.evaluators.items():
                if multi_kwargs:
                    # multi-evaluator - get appropriate runtime kwargs if present
                    kwargs = (
                        eval_kwargs_lists[name] if name in eval_kwargs_lists else {}
                    )
                else:
                    # single evaluator (maintain backwards compatibility)
                    kwargs = eval_kwargs_lists
                eval_kwargs = self._get_eval_kwargs(kwargs, idx)
                eval_jobs.append(
                    eval_response_worker(
                        self.semaphore,
                        evaluator,
                        name,
                        query=query,
                        response=response,
                        eval_kwargs=eval_kwargs,
                    )
                )
        results = await self.asyncio_mod.gather(*eval_jobs)

        # Format results
        return self._format_results(results)

    async def aevaluate_queries(
        self,
        query_engine: BaseQueryEngine,
        queries: Optional[List[str]] = None,
        **eval_kwargs_lists: Dict[str, Any],
    ) -> Dict[str, List[EvaluationResult]]:
        """
        Evaluate queries.

        Args:
            query_engine (BaseQueryEngine): Query engine.
            queries (Optional[List[str]]): List of query strings. Defaults to None.
            **eval_kwargs_lists (Dict[str, Any]): Dict of lists of kwargs to
                pass to evaluator. Defaults to None.

        """
        if queries is None:
            raise ValueError("`queries` must be provided")

        # gather responses
        response_jobs = []
        for query in queries:
            response_jobs.append(response_worker(self.semaphore, query_engine, query))
        responses = await self.asyncio_mod.gather(*response_jobs)

        return await self.aevaluate_responses(
            queries=queries,
            responses=responses,
            **eval_kwargs_lists,
        )

    def evaluate_response_strs(
        self,
        queries: Optional[List[str]] = None,
        response_strs: Optional[List[str]] = None,
        contexts_list: Optional[List[List[str]]] = None,
        **eval_kwargs_lists: Dict[str, Any],
    ) -> Dict[str, List[EvaluationResult]]:
        """
        Evaluate query, response pairs.

        Sync version of aevaluate_response_strs.

        """
        return asyncio_run(
            self.aevaluate_response_strs(
                queries=queries,
                response_strs=response_strs,
                contexts_list=contexts_list,
                **eval_kwargs_lists,
            )
        )

    def evaluate_responses(
        self,
        queries: Optional[List[str]] = None,
        responses: Optional[List[Response]] = None,
        **eval_kwargs_lists: Dict[str, Any],
    ) -> Dict[str, List[EvaluationResult]]:
        """
        Evaluate query, response objs.

        Sync version of aevaluate_responses.

        """
        return asyncio_run(
            self.aevaluate_responses(
                queries=queries,
                responses=responses,
                **eval_kwargs_lists,
            )
        )

    def evaluate_queries(
        self,
        query_engine: BaseQueryEngine,
        queries: Optional[List[str]] = None,
        **eval_kwargs_lists: Dict[str, Any],
    ) -> Dict[str, List[EvaluationResult]]:
        """
        Evaluate queries.

        Sync version of aevaluate_queries.

        """
        return asyncio_run(
            self.aevaluate_queries(
                query_engine=query_engine,
                queries=queries,
                **eval_kwargs_lists,
            )
        )

    def upload_eval_results(
        self,
        project_name: str,
        app_name: str,
        results: Dict[str, List[EvaluationResult]],
    ) -> None:
        """
        Upload the evaluation results to LlamaCloud.

        Args:
            project_name (str): The name of the project.
            app_name (str): The name of the app.
            results (Dict[str, List[EvaluationResult]]):
                The evaluation results, a mapping of metric name to a list of EvaluationResult objects.

        Examples:
            ```python
            results = batch_runner.evaluate_responses(...)

            batch_runner.upload_eval_results(
                project_name="my_project",
                app_name="my_app",
                results=results
            )
            ```

        """
        from llama_index.core.evaluation.eval_utils import upload_eval_results

        upload_eval_results(
            project_name=project_name, app_name=app_name, results=results
        )

aevaluate_response_strs async

aevaluate_response_strs(queries: Optional[List[str]] = None, response_strs: Optional[List[str]] = None, contexts_list: Optional[List[List[str]]] = None, **eval_kwargs_lists: Dict[str, Any]) -> Dict[str, List[EvaluationResult]]

评估查询、响应对。

这将以字符串输入的形式评估查询、响应和上下文。可以在 eval_kwargs_lists 中向评估器提供额外的 kwargs。

参数

名称	类型	描述	默认值
queries	Optional[List[str]]	查询字符串列表。默认为 None。	无
response_strs	Optional[List[str]]	响应字符串列表。默认为 None。	无
contexts_list	Optional[List[List[str]]]	上下文列表的列表。默认为 None。	无
**eval_kwargs_lists	Dict[str, Any]	用于传递给评估器的 kwargs 字典或列表的字典。默认为 None。多个评估器: {评估器: {kwarg: [值列表]},...} 单个评估器: {kwarg: [值列表]}	{}

源代码位于 llama-index-core/llama_index/core/evaluation/batch_runner.py

async def aevaluate_response_strs(
    self,
    queries: Optional[List[str]] = None,
    response_strs: Optional[List[str]] = None,
    contexts_list: Optional[List[List[str]]] = None,
    **eval_kwargs_lists: Dict[str, Any],
) -> Dict[str, List[EvaluationResult]]:
    """
    Evaluate query, response pairs.

    This evaluates queries, responses, contexts as string inputs.
    Can supply additional kwargs to the evaluator in eval_kwargs_lists.

    Args:
        queries (Optional[List[str]]): List of query strings. Defaults to None.
        response_strs (Optional[List[str]]): List of response strings.
            Defaults to None.
        contexts_list (Optional[List[List[str]]]): List of context lists.
            Defaults to None.
        **eval_kwargs_lists (Dict[str, Any]): Dict of either dicts or lists
            of kwargs to pass to evaluator. Defaults to None.
                multiple evaluators: {evaluator: {kwarg: [list of values]},...}
                single evaluator:    {kwarg: [list of values]}

    """
    queries, response_strs, contexts_list = self._validate_and_clean_inputs(
        queries, response_strs, contexts_list
    )
    eval_kwargs_lists = self._validate_nested_eval_kwargs_types(eval_kwargs_lists)

    # boolean to check if using multi kwarg evaluator
    multi_kwargs = len(eval_kwargs_lists) > 0 and isinstance(
        next(iter(eval_kwargs_lists.values())), dict
    )

    # run evaluations
    eval_jobs = []
    for idx, query in enumerate(cast(List[str], queries)):
        response_str = cast(List, response_strs)[idx]
        contexts = cast(List, contexts_list)[idx]
        for name, evaluator in self.evaluators.items():
            if multi_kwargs:
                # multi-evaluator - get appropriate runtime kwargs if present
                kwargs = (
                    eval_kwargs_lists[name] if name in eval_kwargs_lists else {}
                )
            else:
                # single evaluator (maintain backwards compatibility)
                kwargs = eval_kwargs_lists
            eval_kwargs = self._get_eval_kwargs(kwargs, idx)
            eval_jobs.append(
                eval_worker(
                    self.semaphore,
                    evaluator,
                    name,
                    query=query,
                    response_str=response_str,
                    contexts=contexts,
                    eval_kwargs=eval_kwargs,
                )
            )
    results = await self.asyncio_mod.gather(*eval_jobs)

    # Format results
    return self._format_results(results)

aevaluate_responses async

aevaluate_responses(queries: Optional[List[str]] = None, responses: Optional[List[Response]] = None, **eval_kwargs_lists: Dict[str, Any]) -> Dict[str, List[EvaluationResult]]

评估查询、响应对。

这将评估查询和响应对象。

参数

名称	类型	描述	默认值
queries	Optional[List[str]]	查询字符串列表。默认为 None。	无
responses	Optional[List[Response]]	响应对象列表。默认为 None。	无
**eval_kwargs_lists	Dict[str, Any]	用于传递给评估器的 kwargs 字典或列表的字典。默认为 None。多个评估器: {评估器: {kwarg: [值列表]},...} 单个评估器: {kwarg: [值列表]}	{}

源代码位于 llama-index-core/llama_index/core/evaluation/batch_runner.py

async def aevaluate_responses(
    self,
    queries: Optional[List[str]] = None,
    responses: Optional[List[Response]] = None,
    **eval_kwargs_lists: Dict[str, Any],
) -> Dict[str, List[EvaluationResult]]:
    """
    Evaluate query, response pairs.

    This evaluates queries and response objects.

    Args:
        queries (Optional[List[str]]): List of query strings. Defaults to None.
        responses (Optional[List[Response]]): List of response objects.
            Defaults to None.
        **eval_kwargs_lists (Dict[str, Any]): Dict of either dicts or lists
            of kwargs to pass to evaluator. Defaults to None.
                multiple evaluators: {evaluator: {kwarg: [list of values]},...}
                single evaluator:    {kwarg: [list of values]}

    """
    queries, responses = self._validate_and_clean_inputs(queries, responses)
    eval_kwargs_lists = self._validate_nested_eval_kwargs_types(eval_kwargs_lists)

    # boolean to check if using multi kwarg evaluator
    multi_kwargs = len(eval_kwargs_lists) > 0 and isinstance(
        next(iter(eval_kwargs_lists.values())), dict
    )

    # run evaluations
    eval_jobs = []
    for idx, query in enumerate(cast(List[str], queries)):
        response = cast(List, responses)[idx]
        for name, evaluator in self.evaluators.items():
            if multi_kwargs:
                # multi-evaluator - get appropriate runtime kwargs if present
                kwargs = (
                    eval_kwargs_lists[name] if name in eval_kwargs_lists else {}
                )
            else:
                # single evaluator (maintain backwards compatibility)
                kwargs = eval_kwargs_lists
            eval_kwargs = self._get_eval_kwargs(kwargs, idx)
            eval_jobs.append(
                eval_response_worker(
                    self.semaphore,
                    evaluator,
                    name,
                    query=query,
                    response=response,
                    eval_kwargs=eval_kwargs,
                )
            )
    results = await self.asyncio_mod.gather(*eval_jobs)

    # Format results
    return self._format_results(results)

aevaluate_queries async

aevaluate_queries(query_engine: BaseQueryEngine, queries: Optional[List[str]] = None, **eval_kwargs_lists: Dict[str, Any]) -> Dict[str, List[EvaluationResult]]

评估查询。

参数

名称	类型	描述	默认值
query_engine	BaseQueryEngine	查询引擎。	required
queries	Optional[List[str]]	查询字符串列表。默认为 None。	无
**eval_kwargs_lists	Dict[str, Any]	用于传递给评估器的 kwargs 列表字典。默认为 None。	{}

源代码位于 llama-index-core/llama_index/core/evaluation/batch_runner.py

async def aevaluate_queries(
    self,
    query_engine: BaseQueryEngine,
    queries: Optional[List[str]] = None,
    **eval_kwargs_lists: Dict[str, Any],
) -> Dict[str, List[EvaluationResult]]:
    """
    Evaluate queries.

    Args:
        query_engine (BaseQueryEngine): Query engine.
        queries (Optional[List[str]]): List of query strings. Defaults to None.
        **eval_kwargs_lists (Dict[str, Any]): Dict of lists of kwargs to
            pass to evaluator. Defaults to None.

    """
    if queries is None:
        raise ValueError("`queries` must be provided")

    # gather responses
    response_jobs = []
    for query in queries:
        response_jobs.append(response_worker(self.semaphore, query_engine, query))
    responses = await self.asyncio_mod.gather(*response_jobs)

    return await self.aevaluate_responses(
        queries=queries,
        responses=responses,
        **eval_kwargs_lists,
    )

evaluate_response_strs

evaluate_response_strs(queries: Optional[List[str]] = None, response_strs: Optional[List[str]] = None, contexts_list: Optional[List[List[str]]] = None, **eval_kwargs_lists: Dict[str, Any]) -> Dict[str, List[EvaluationResult]]

评估查询、响应对。

aevaluate_response_strs 的同步版本。

源代码位于 llama-index-core/llama_index/core/evaluation/batch_runner.py

def evaluate_response_strs(
    self,
    queries: Optional[List[str]] = None,
    response_strs: Optional[List[str]] = None,
    contexts_list: Optional[List[List[str]]] = None,
    **eval_kwargs_lists: Dict[str, Any],
) -> Dict[str, List[EvaluationResult]]:
    """
    Evaluate query, response pairs.

    Sync version of aevaluate_response_strs.

    """
    return asyncio_run(
        self.aevaluate_response_strs(
            queries=queries,
            response_strs=response_strs,
            contexts_list=contexts_list,
            **eval_kwargs_lists,
        )
    )

evaluate_responses

evaluate_responses(queries: Optional[List[str]] = None, responses: Optional[List[Response]] = None, **eval_kwargs_lists: Dict[str, Any]) -> Dict[str, List[EvaluationResult]]

评估查询、响应对象。

aevaluate_responses 的同步版本。

源代码位于 llama-index-core/llama_index/core/evaluation/batch_runner.py

def evaluate_responses(
    self,
    queries: Optional[List[str]] = None,
    responses: Optional[List[Response]] = None,
    **eval_kwargs_lists: Dict[str, Any],
) -> Dict[str, List[EvaluationResult]]:
    """
    Evaluate query, response objs.

    Sync version of aevaluate_responses.

    """
    return asyncio_run(
        self.aevaluate_responses(
            queries=queries,
            responses=responses,
            **eval_kwargs_lists,
        )
    )

evaluate_queries

evaluate_queries(query_engine: BaseQueryEngine, queries: Optional[List[str]] = None, **eval_kwargs_lists: Dict[str, Any]) -> Dict[str, List[EvaluationResult]]

评估查询。

aevaluate_queries 的同步版本。

源代码位于 llama-index-core/llama_index/core/evaluation/batch_runner.py

def evaluate_queries(
    self,
    query_engine: BaseQueryEngine,
    queries: Optional[List[str]] = None,
    **eval_kwargs_lists: Dict[str, Any],
) -> Dict[str, List[EvaluationResult]]:
    """
    Evaluate queries.

    Sync version of aevaluate_queries.

    """
    return asyncio_run(
        self.aevaluate_queries(
            query_engine=query_engine,
            queries=queries,
            **eval_kwargs_lists,
        )
    )

upload_eval_results

upload_eval_results(project_name: str, app_name: str, results: Dict[str, List[EvaluationResult]]) -> None

将评估结果上传到 LlamaCloud。

参数

名称	类型	描述	默认值
project_name	str	项目名称。	required
app_name	str	应用程序名称。	required
results	Dict[str, List[EvaluationResult]]	评估结果，按指标名称映射到 EvaluationResult 对象列表。	required

示例

results = batch_runner.evaluate_responses(...)

batch_runner.upload_eval_results(
    project_name="my_project",
    app_name="my_app",
    results=results
)

源代码位于 llama-index-core/llama_index/core/evaluation/batch_runner.py

def upload_eval_results(
    self,
    project_name: str,
    app_name: str,
    results: Dict[str, List[EvaluationResult]],
) -> None:
    """
    Upload the evaluation results to LlamaCloud.

    Args:
        project_name (str): The name of the project.
        app_name (str): The name of the app.
        results (Dict[str, List[EvaluationResult]]):
            The evaluation results, a mapping of metric name to a list of EvaluationResult objects.

    Examples:
        ```python
        results = batch_runner.evaluate_responses(...)

        batch_runner.upload_eval_results(
            project_name="my_project",
            app_name="my_app",
            results=results
        )
        ```

    """
    from llama_index.core.evaluation.eval_utils import upload_eval_results

    upload_eval_results(
        project_name=project_name, app_name=app_name, results=results
    )