simple

libs/langchain/langchain/evaluation/parsing/json_distance.py

@@ -0,0 +1,66 @@
+import json
+from typing import Any, Callable, Optional, Union
+
+from langchain.evaluation.schema import StringEvaluator
+from langchain.output_parsers.json import parse_json_markdown
+
+
+class JsonEditDistanceEvaluator(StringEvaluator):
+    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  # noqa: F401
+            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 _distance(self, a: Any, b: Any) -> float:
+        return self._string_distance(a, b)
+
+    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._distance(parsed, label)
+        return {"score": distance}

libs/langchain/tests/unit_tests/evaluation/parsing/test_json_distance.py

@@ -0,0 +1,107 @@
+import pytest
+
+from langchain.evaluation.parsing.json_distance import JsonEditDistanceEvaluator
+
+
+@pytest.fixture
+def json_distance_evaluator() -> JsonEditDistanceEvaluator:
+    return JsonEditDistanceEvaluator()
+
+
+def test_json_distance_evaluator_requires_input(
+    json_distance_evaluator: JsonEditDistanceEvaluator,
+) -> None:
+    assert json_distance_evaluator.requires_input is False
+
+
+def test_json_distance_evaluator_requires_reference(
+    json_distance_evaluator: JsonEditDistanceEvaluator,
+) -> None:
+    assert json_distance_evaluator.requires_reference is True
+
+
+def test_json_distance_evaluator_evaluation_name(
+    json_distance_evaluator: JsonEditDistanceEvaluator,
+) -> None:
+    assert json_distance_evaluator.evaluation_name == "json_edit_distance"
+
+
+def test_json_distance_evaluator_parse_json(
+    json_distance_evaluator: JsonEditDistanceEvaluator,
+) -> None:
+    string = '{"a": 1}'
+    result = json_distance_evaluator._parse_json(string)
+    assert result == {"a": 1}
+
+
+def test_json_distance_evaluator_evaluate_strings_simple_diff(
+    json_distance_evaluator: JsonEditDistanceEvaluator,
+) -> None:
+    prediction = '{"a":           1}'
+    reference = '{"a": 2}'
+    result = json_distance_evaluator._evaluate_strings(
+        prediction=prediction, reference=reference
+    )
+    # Only 1 character flipped
+    pytest.approx(1 / 7, result["score"])
+
+
+def test_json_distance_evaluator_evaluate_strings_complex_diff(
+    json_distance_evaluator: JsonEditDistanceEvaluator,
+) -> None:
+    prediction = '{"a":1, "b": {"c": 2, "d": 3}}'
+    reference = '{"a": 1, "b": {"c": 2, "d": 4}}'
+    result = json_distance_evaluator._evaluate_strings(
+        prediction=prediction, reference=reference
+    )
+    # Only 1 character flipped
+    pytest.approx(1 / len(reference.replace(" ", "")), result["score"])
+
+
+def test_json_distance_evaluator_evaluate_strings_list_diff(
+    json_distance_evaluator: JsonEditDistanceEvaluator,
+) -> None:
+    prediction = '[{"a": 1, "b": 2}, {"a": 2, "b": 3}]'
+    reference = '[{"a": 1, "b": 2}, {"a": 2, "b": 4}]'
+    result = json_distance_evaluator._evaluate_strings(
+        prediction=prediction, reference=reference
+    )
+    # Again only 1 character flipped
+    pytest.approx(1 / len(reference.replace(" ", "")), result["score"])
+
+
+def test_json_distance_evaluator_evaluate_strings_list_same(
+    json_distance_evaluator: JsonEditDistanceEvaluator,
+) -> None:
+    prediction = '[{"a": 1, "b": 2}, {"a": 2, "b": 3}]'
+    reference = '[{"b": 2, "a": 1}, {"b": 3, "a": 2}]'
+    result = json_distance_evaluator._evaluate_strings(
+        prediction=prediction, reference=reference
+    )
+    assert result["score"] == 0
+
+
+def test_json_distance_evaluator_evaluate_strings_list_diff_length(
+    json_distance_evaluator: JsonEditDistanceEvaluator,
+) -> None:
+    prediction = '[{"a": 1, "b": 2}, {"a": 2, "b": 3}]'
+    reference = '[{"a": 1, "b": 2}]'
+    result = json_distance_evaluator._evaluate_strings(
+        prediction=prediction, reference=reference
+    )
+    pytest.approx(
+        len('{"a":2,"b":3}') / len(reference.replace(" ", "")), result["score"]
+    )
+
+
+def test_json_distance_evaluator_evaluate_strings_custom_operator_equal() -> None:
+    """Custom operator that returns 0.5 if strings are different."""
+
+    def custom_distance(a: str, b: str) -> float:
+        return 0.5 if a != b else 0.0
+
+    evaluator = JsonEditDistanceEvaluator(string_distance=custom_distance)
+    prediction = '{"a": "apple", "b": "banana"}'
+    reference = '{"a": "apple", "b": "berries"}'
+    result = evaluator._evaluate_strings(prediction=prediction, reference=reference)
+    assert result["score"] == 0.5