Source code for langchain.evaluation.parsing.json_distance

import json
from typing import Any, Callable, Optional, Union

from langchain_core.utils.json import parse_json_markdown

from langchain.evaluation.schema import StringEvaluator


class JsonEditDistanceEvaluator(StringEvaluator):
    """一个计算JSON字符串之间编辑距离的评估器。

该评估器在解析并将两个JSON字符串转换为规范格式（即，空格和键的顺序被规范化）之后，计算两个JSON字符串之间的标准化Damerau-Levenshtein距离。它可以通过替代距离和规范化函数进行定制。

参数：
    string_distance（可选[Callable[[str, str], float]]）：计算两个字符串之间距离的可调用函数。
        如果未提供，则将使用`rapidfuzz`包中的Damerau-Levenshtein距离。
    canonicalize（可选[Callable[[Any], Any]]）：将解析的JSON对象转换为其规范字符串形式的可调用函数。
        如果未提供，默认行为是使用排序后的键和没有额外空格序列化JSON。
    **kwargs（Any）：额外的关键字参数。

属性：
    _string_distance（Callable[[str, str], float]）：内部距离计算函数。
    _canonicalize（Callable[[Any], Any]）：内部规范化函数。

示例：
    >>> evaluator = JsonEditDistanceEvaluator()
    >>> result = evaluator.evaluate_strings(prediction='{"a": 1, "b": 2}', reference='{"a": 1, "b": 3}')
    >>> assert result["score"] is not None

引发：
    ImportError：如果未安装`rapidfuzz`并且未提供替代的`string_distance`函数。"""  # noqa: E501

    def __init__(
        self,
        string_distance: Optional[Callable[[str, str], float]] = None,
        canonicalize: Optional[Callable[[Any], Any]] = None,
        **kwargs: Any,
    ) -> None:
        super().__init__()
        if string_distance is not None:
            self._string_distance = string_distance
        else:
            try:
                from rapidfuzz import distance as rfd
            except ImportError:
                raise ImportError(
                    "The default string_distance operator for the "
                    " JsonEditDistanceEvaluator requires installation of "
                    "the rapidfuzz package. "
                    "Please install it with `pip install rapidfuzz`."
                )
            self._string_distance = rfd.DamerauLevenshtein.normalized_distance
        if canonicalize is not None:
            self._canonicalize = canonicalize
        else:
            self._canonicalize = lambda x: json.dumps(
                x,
                separators=(",", ":"),
                sort_keys=True,  # eliminate whitespace
            )

    @property
    def requires_input(self) -> bool:
        return False

    @property
    def requires_reference(self) -> bool:
        return True

    @property
    def evaluation_name(self) -> str:
        return "json_edit_distance"

    def _parse_json(self, node: Any) -> Union[dict, list, None, float, bool, int, str]:
        if isinstance(node, str):
            return parse_json_markdown(node)
        return node

    def _evaluate_strings(
        self,
        prediction: str,
        input: Optional[str] = None,
        reference: Optional[str] = None,
        **kwargs: Any,
    ) -> dict:
        parsed = self._canonicalize(self._parse_json(prediction))
        label = self._canonicalize(self._parse_json(reference))
        distance = self._string_distance(parsed, label)
        return {"score": distance}