_RapidFuzzChainMixin Class — langchain Architecture

Source Code

libs/langchain/langchain_classic/evaluation/string_distance/base.py lines 61–162

class _RapidFuzzChainMixin(Chain):
    """Shared methods for the rapidfuzz string distance evaluators."""

    distance: StringDistance = Field(default=StringDistance.JARO_WINKLER)
    normalize_score: bool = Field(default=True)
    """Whether to normalize the score to a value between `0` and `1`.
    Applies only to the Levenshtein and Damerau-Levenshtein distances."""

    @pre_init
    def validate_dependencies(cls, values: dict[str, Any]) -> dict[str, Any]:
        """Validate that the rapidfuzz library is installed.

        Args:
            values: The input values.

        Returns:
            The validated values.
        """
        _load_rapidfuzz()
        return values

    @property
    def output_keys(self) -> list[str]:
        """Get the output keys.

        Returns:
            The output keys.
        """
        return ["score"]

    def _prepare_output(self, result: dict[str, Any]) -> dict[str, Any]:
        """Prepare the output dictionary.

        Args:
            result: The evaluation results.

        Returns:
            The prepared output dictionary.
        """
        result = {"score": result["score"]}
        if RUN_KEY in result:
            result[RUN_KEY] = result[RUN_KEY].dict()
        return result

    @staticmethod
    def _get_metric(distance: str, *, normalize_score: bool = False) -> Callable:
        """Get the distance metric function based on the distance type.

        Args:
            distance: The distance type.
            normalize_score: Whether to normalize the score.

        Returns:
            The distance metric function.

        Raises:
            ValueError: If the distance metric is invalid.
        """
        from rapidfuzz import distance as rf_distance

        module_map: dict[str, Any] = {
            StringDistance.DAMERAU_LEVENSHTEIN: rf_distance.DamerauLevenshtein,
            StringDistance.LEVENSHTEIN: rf_distance.Levenshtein,
            StringDistance.JARO: rf_distance.Jaro,
            StringDistance.JARO_WINKLER: rf_distance.JaroWinkler,
            StringDistance.HAMMING: rf_distance.Hamming,
            StringDistance.INDEL: rf_distance.Indel,
        }
        if distance not in module_map:
            msg = (
                f"Invalid distance metric: {distance}"
                f"\nMust be one of: {list(StringDistance)}"
            )
            raise ValueError(msg)
        module = module_map[distance]
        if normalize_score:
            return module.normalized_distance
        return module.distance

    @property
    def metric(self) -> Callable: