feat(checks): add consensus retries for LLM-based checks

libs/giskard-checks/src/giskard/checks/judges/base.py

@@ -1,4 +1,5 @@
-from typing import Any, override
+from collections import Counter
+from typing import Any, Literal, override
 
 from giskard.agents.templates import MessageTemplate
 from giskard.agents.workflow import ChatWorkflow, TemplateReference
@@ -34,8 +35,23 @@ class BaseLLMCheck[InputType, OutputType, TraceType: Trace](  # pyright: ignore[
     generator : BaseGenerator
         Generator for LLM evaluation. Defaults to the global
         default generator if not specified.
+    num_runs : int
+        Number of times to execute the LLM-based evaluation.
+    consensus : Literal["majority", "unanimous", "any"]
+        Strategy used to aggregate multiple runs into a final result.
     """
 
+    num_runs: int = Field(
+        default=1,
+        ge=1,
+        strict=True,
+        description="Number of times to execute the LLM evaluation.",
+    )
+    consensus: Literal["majority", "unanimous", "any"] = Field(
+        default="majority",
+        description="Strategy used to aggregate multiple runs into a final result.",
+    )
+
     @property
     def output_type(self) -> type[BaseModel] | None:
         return LLMCheckResult
@@ -91,9 +107,23 @@ async def run(self, trace: TraceType) -> CheckResult:
         CheckResult
             The result of the check evaluation.
         """
-        workflow = await self._build_workflow(trace)
-
         inputs = await self.get_inputs(trace)
+        results = [
+            await self._run_once(trace, inputs=inputs) for _ in range(self.num_runs)
+        ]
+
+        if self.num_runs == 1:
+            return results[0]
+
+        return self._aggregate_results(results)
+
+    async def _run_once(
+        self,
+        trace: TraceType,
+        *,
+        inputs: dict[str, Any],
+    ) -> CheckResult:
+        workflow = await self._build_workflow(trace)
         workflow = workflow.with_inputs(**inputs)
 
         if self.output_type is not None:
@@ -103,6 +133,53 @@ async def run(self, trace: TraceType) -> CheckResult:
 
         return await self._handle_output(chat.output, inputs, trace)
 
+    def _aggregate_results(self, results: list[CheckResult]) -> CheckResult:
+        passed_count = sum(result.passed for result in results)
+        non_pass_count = len(results) - passed_count
+
+        if self.consensus == "unanimous":
+            consensus_passed = passed_count == len(results)
+        elif self.consensus == "any":
+            consensus_passed = passed_count >= 1
+        else:
+            consensus_passed = passed_count > non_pass_count
+
+        representative = (
+            self._select_pass_result(results)
+            if consensus_passed
+            else self._select_non_pass_result(results)
+        )
+        status_counts = Counter(result.status.value for result in results)
+
+        return representative.model_copy(
+            update={
+                "details": {
+                    **representative.details,
+                    "runs": results,
+                    "num_runs": self.num_runs,
+                    "consensus": self.consensus,
+                    "consensus_passed": consensus_passed,
+                    "status_counts": dict(status_counts),
+                }
+            }
+        )
+
+    @staticmethod
+    def _select_pass_result(results: list[CheckResult]) -> CheckResult:
+        return next((result for result in results if result.passed), results[0])
+
+    @staticmethod
+    def _select_non_pass_result(results: list[CheckResult]) -> CheckResult:
+        for predicate in (
+            lambda result: result.failed,
+            lambda result: result.errored,
+            lambda result: result.skipped,
+        ):
+            match = next((result for result in results if predicate(result)), None)
+            if match is not None:
+                return match
+        return results[0]
+
     async def get_inputs(self, trace: TraceType) -> dict[str, Any]:
         """Get template inputs for the LLM prompt.
 

libs/giskard-checks/tests/builtin/test_base.py

@@ -42,7 +42,130 @@ async def _call_model(
         )
 
 
+class SequenceMockGenerator(BaseGenerator):
+    responses: list[tuple[bool, str | None]]
+    calls: list[Sequence[ChatMessage]] = Field(default_factory=list)
+
+    @override
+    async def _call_model(
+        self,
+        messages: Sequence[ChatMessage],
+        params: GenerationParams,
+        metadata: dict[str, Any] | None = None,
+    ) -> CompletionResponse:
+        self.calls.append(messages)
+        passed, reason = self.responses[len(self.calls) - 1]
+        return CompletionResponse(
+            choices=[
+                Choice(
+                    message=AssistantMessage(
+                        content=json.dumps(
+                            {
+                                "passed": passed,
+                                "reason": reason,
+                            }
+                        )
+                    ),
+                    finish_reason="stop",
+                    index=0,
+                )
+            ]
+        )
+
+
 class TestBaseLLMCheck:
+    async def test_majority_consensus_returns_pass_with_run_details(self):
+        class ConsensusCheck(BaseLLMCheck[str, str, Trace[str, str]]):
+            @override
+            def get_prompt(self) -> str:
+                return "Evaluate."
+
+        generator = SequenceMockGenerator(
+            responses=[
+                (True, "pass-1"),
+                (False, "fail-2"),
+                (True, "pass-3"),
+            ]
+        )
+        check = ConsensusCheck(generator=generator, num_runs=3, consensus="majority")
+
+        result = await check.run(Trace())
+
+        assert result.passed
+        assert result.message == "pass-1"
+        assert result.details["consensus"] == "majority"
+        assert result.details["num_runs"] == 3
+        assert result.details["consensus_passed"] is True
+        assert result.details["status_counts"] == {"pass": 2, "fail": 1}
+        assert [run.status for run in result.details["runs"]] == [
+            "pass",
+            "fail",
+            "pass",
+        ]
+        assert len(generator.calls) == 3
+
+    async def test_unanimous_consensus_requires_all_runs_to_pass(self):
+        class ConsensusCheck(BaseLLMCheck[str, str, Trace[str, str]]):
+            @override
+            def get_prompt(self) -> str:
+                return "Evaluate."
+
+        generator = SequenceMockGenerator(
+            responses=[
+                (True, "pass-1"),
+                (False, "fail-2"),
+                (True, "pass-3"),
+            ]
+        )
+        check = ConsensusCheck(generator=generator, num_runs=3, consensus="unanimous")
+
+        result = await check.run(Trace())
+
+        assert result.failed
+        assert result.message == "fail-2"
+        assert result.details["consensus"] == "unanimous"
+        assert result.details["consensus_passed"] is False
+
+    async def test_any_consensus_passes_if_any_run_passes(self):
+        class ConsensusCheck(BaseLLMCheck[str, str, Trace[str, str]]):
+            @override
+            def get_prompt(self) -> str:
+                return "Evaluate."
+
+        generator = SequenceMockGenerator(
+            responses=[
+                (False, "fail-1"),
+                (False, "fail-2"),
+                (True, "pass-3"),
+            ]
+        )
+        check = ConsensusCheck(generator=generator, num_runs=3, consensus="any")
+
+        result = await check.run(Trace())
+
+        assert result.passed
+        assert result.message == "pass-3"
+        assert result.details["consensus"] == "any"
+        assert result.details["consensus_passed"] is True
+
+    async def test_single_run_keeps_default_behavior_unchanged(self):
+        class ConsensusCheck(BaseLLMCheck[str, str, Trace[str, str]]):
+            @override
+            def get_prompt(self) -> str:
+                return "Evaluate."
+
+        generator = SequenceMockGenerator(responses=[(True, "pass-1")])
+        check = ConsensusCheck(generator=generator)
+
+        result = await check.run(Trace())
+
+        assert result.passed
+        assert result.details == {
+            "reason": "pass-1",
+            "inputs": {"trace": Trace()},
+        }
+        assert "runs" not in result.details
+
     async def test_custom_output_type_requires_handle_output(self):
         class CustomOutputType(BaseModel):
             score: float