lint

Author: A-Vamshi

deepeval/optimizer/algorithms/gepa/gepa.py

@@ -1 +1 @@
-from .miprov2 import MIPROV2
+from .miprov2 import MIPROV2

deepeval/optimizer/algorithms/miprov2/__init__.py

@@ -19,6 +19,7 @@
 @dataclass
 class Demonstration:
     """A single, mathematically verified few-shot example."""
+
     input_text: str
     output_text: str
     golden_index: int = -1
@@ -27,6 +28,7 @@ class Demonstration:
 @dataclass
 class DemonstrationSet:
     """A set of demonstrations to be dynamically injected into a prompt."""
+
     demonstrations: List[Demonstration] = field(default_factory=list)
     id: str = ""
 
@@ -37,10 +39,11 @@ def __post_init__(self):
     def to_text(self, max_demonstrations: Optional[int] = None) -> str:
         """Render demonstrations as text for inclusion in prompts."""
         demos_to_use = (
-            self.demonstrations[:max_demonstrations] 
-            if max_demonstrations else self.demonstrations
+            self.demonstrations[:max_demonstrations]
+            if max_demonstrations
+            else self.demonstrations
         )
-        
+
         if not demos_to_use:
             return ""
 
@@ -56,7 +59,7 @@ def to_text(self, max_demonstrations: Optional[int] = None) -> str:
 
 class DemonstrationBootstrapper:
     """
-    Bootstraps few-shot demonstrations by running the prompt on training 
+    Bootstraps few-shot demonstrations by running the prompt on training
     examples and keeping strictly successful outputs based on metric success.
     """
 
@@ -78,28 +81,42 @@ def __init__(
         else:
             self.random_state = random_state or random.Random()
 
-    def _extract_input(self, golden: Union[Golden, ConversationalGolden]) -> str:
+    def _extract_input(
+        self, golden: Union[Golden, ConversationalGolden]
+    ) -> str:
         """Strictly extract the input text, throwing errors on invalid state."""
         if isinstance(golden, Golden):
             if not golden.input:
-                raise DeepEvalError("Golden must have a valid 'input' for MIPROv2 bootstrapping.")
+                raise DeepEvalError(
+                    "Golden must have a valid 'input' for MIPROv2 bootstrapping."
+                )
             return golden.input
-            
+
         else:
-            user_turns = [t.content for t in (golden.turns or []) if t.role == "user"]
+            user_turns = [
+                t.content for t in (golden.turns or []) if t.role == "user"
+            ]
             if not user_turns:
-                raise DeepEvalError("ConversationalGolden must have at least one 'user' turn for MIPROv2 bootstrapping.")
+                raise DeepEvalError(
+                    "ConversationalGolden must have at least one 'user' turn for MIPROv2 bootstrapping."
+                )
             return "\n".join(user_turns)
 
-    def _extract_expected_output(self, golden: Union[Golden, ConversationalGolden]) -> Optional[str]:
+    def _extract_expected_output(
+        self, golden: Union[Golden, ConversationalGolden]
+    ) -> Optional[str]:
         """Strictly extract the expected output/outcome if it exists."""
         if isinstance(golden, Golden):
             if not golden.expected_output:
-                raise DeepEvalError("Golden must have a valid 'expected_output' for MIPROv2 bootstrapping.")
+                raise DeepEvalError(
+                    "Golden must have a valid 'expected_output' for MIPROv2 bootstrapping."
+                )
             return str(golden.expected_output)
         else:
             if not golden.expected_outcome:
-                raise DeepEvalError("ConversationalGolden must have a valid 'expected_outcome' for MIPROv2 bootstrapping.")
+                raise DeepEvalError(
+                    "ConversationalGolden must have a valid 'expected_outcome' for MIPROv2 bootstrapping."
+                )
             return golden.expected_outcome
 
     def bootstrap(
@@ -114,25 +131,43 @@ def bootstrap(
         shuffled_indices = list(range(len(goldens)))
         self.random_state.shuffle(shuffled_indices)
 
-        max_attempts = min(len(goldens), self.max_bootstrapped_demonstrations * 3)
+        max_attempts = min(
+            len(goldens), self.max_bootstrapped_demonstrations * 3
+        )
         prompt_dict = {"__module__": prompt}
 
         for idx in shuffled_indices[:max_attempts]:
             golden = goldens[idx]
             input_text = self._extract_input(golden)
             expected = self._extract_expected_output(golden)
 
-            if expected and len(labeled_demonstrations) < self.max_labeled_demonstrations * self.num_demonstration_sets:
-                labeled_demonstrations.append(Demonstration(input_text=input_text, output_text=expected, golden_index=idx))
-
-            if len(all_demonstrations) < self.max_bootstrapped_demonstrations * self.num_demonstration_sets:
+            if (
+                expected
+                and len(labeled_demonstrations)
+                < self.max_labeled_demonstrations * self.num_demonstration_sets
+            ):
+                labeled_demonstrations.append(
+                    Demonstration(
+                        input_text=input_text,
+                        output_text=expected,
+                        golden_index=idx,
+                    )
+                )
+
+            if (
+                len(all_demonstrations)
+                < self.max_bootstrapped_demonstrations
+                * self.num_demonstration_sets
+            ):
                 try:
                     # 1. Generate actual output
                     actual_output = self.scorer.generate(prompt_dict, golden)
-                    
+
                     # 2. Build the test case safely
-                    test_case = self.scorer._golden_to_test_case(golden, actual_output)
-                
+                    test_case = self.scorer._golden_to_test_case(
+                        golden, actual_output
+                    )
+
                     # 3. Evaluate against all metrics
                     metrics = copy_metrics(self.scorer.metrics)
                     is_successful = True
@@ -141,19 +176,32 @@ def bootstrap(
                         if not metric.is_successful():
                             is_successful = False
                             break
-                    
+
                     # 4. Save if all metrics passed
                     if is_successful:
-                        all_demonstrations.append(Demonstration(input_text=input_text, output_text=actual_output, golden_index=idx))
+                        all_demonstrations.append(
+                            Demonstration(
+                                input_text=input_text,
+                                output_text=actual_output,
+                                golden_index=idx,
+                            )
+                        )
                 except Exception:
                     continue
 
-            if (len(all_demonstrations) >= self.max_bootstrapped_demonstrations * self.num_demonstration_sets and 
-                len(labeled_demonstrations) >= self.max_labeled_demonstrations * self.num_demonstration_sets):
+            if (
+                len(all_demonstrations)
+                >= self.max_bootstrapped_demonstrations
+                * self.num_demonstration_sets
+                and len(labeled_demonstrations)
+                >= self.max_labeled_demonstrations * self.num_demonstration_sets
+            ):
                 break
 
-        demo_sets = self._create_demonstration_sets(all_demonstrations, labeled_demonstrations)
-        
+        demo_sets = self._create_demonstration_sets(
+            all_demonstrations, labeled_demonstrations
+        )
+
         if not demo_sets or all(not ds.demonstrations for ds in demo_sets):
             raise DeepEvalError(
                 "Bootstrapper failed to generate any demonstrations. "
@@ -172,7 +220,9 @@ async def a_bootstrap(
         shuffled_indices = list(range(len(goldens)))
         self.random_state.shuffle(shuffled_indices)
 
-        max_attempts = min(len(goldens), self.max_bootstrapped_demonstrations * 3)
+        max_attempts = min(
+            len(goldens), self.max_bootstrapped_demonstrations * 3
+        )
         selected_indices = shuffled_indices[:max_attempts]
 
         tasks_info: List[Tuple[int, str, Optional[str]]] = []
@@ -183,23 +233,41 @@ async def a_bootstrap(
             input_text = self._extract_input(golden)
             expected = self._extract_expected_output(golden)
 
-            if expected and len(labeled_demonstrations) < self.max_labeled_demonstrations * self.num_demonstration_sets:
-                labeled_demonstrations.append(Demonstration(input_text=input_text, output_text=expected, golden_index=idx))
+            if (
+                expected
+                and len(labeled_demonstrations)
+                < self.max_labeled_demonstrations * self.num_demonstration_sets
+            ):
+                labeled_demonstrations.append(
+                    Demonstration(
+                        input_text=input_text,
+                        output_text=expected,
+                        golden_index=idx,
+                    )
+                )
 
             tasks_info.append((idx, input_text, expected))
 
-        max_bootstrapped = self.max_bootstrapped_demonstrations * self.num_demonstration_sets
+        max_bootstrapped = (
+            self.max_bootstrapped_demonstrations * self.num_demonstration_sets
+        )
         tasks_info = tasks_info[:max_bootstrapped]
 
-        async def evaluate_one(idx: int, input_text: str, expected: Optional[str]) -> Optional[Demonstration]:
+        async def evaluate_one(
+            idx: int, input_text: str, expected: Optional[str]
+        ) -> Optional[Demonstration]:
             golden = goldens[idx]
             try:
                 # 1. Generate actual output
-                actual_output = await self.scorer.a_generate(prompt_dict, golden)
-                
+                actual_output = await self.scorer.a_generate(
+                    prompt_dict, golden
+                )
+
                 # 2. Build the test case safely
-                test_case = self.scorer._golden_to_test_case(golden, actual_output)
-                
+                test_case = self.scorer._golden_to_test_case(
+                    golden, actual_output
+                )
+
                 # 3. Evaluate against all metrics
                 metrics = copy_metrics(self.scorer.metrics)
                 is_successful = True
@@ -211,16 +279,24 @@ async def evaluate_one(idx: int, input_text: str, expected: Optional[str]) -> Op
 
                 # 4. Save if all metrics passed
                 if is_successful:
-                    return Demonstration(input_text=input_text, output_text=actual_output, golden_index=idx)
+                    return Demonstration(
+                        input_text=input_text,
+                        output_text=actual_output,
+                        golden_index=idx,
+                    )
             except Exception:
                 pass
             return None
 
-        results = await asyncio.gather(*[evaluate_one(idx, inp, exp) for idx, inp, exp in tasks_info])
+        results = await asyncio.gather(
+            *[evaluate_one(idx, inp, exp) for idx, inp, exp in tasks_info]
+        )
         all_demonstrations = [demo for demo in results if demo is not None]
 
-        demo_sets = self._create_demonstration_sets(all_demonstrations, labeled_demonstrations)
-        
+        demo_sets = self._create_demonstration_sets(
+            all_demonstrations, labeled_demonstrations
+        )
+
         if not demo_sets or all(not ds.demonstrations for ds in demo_sets):
             raise DeepEvalError(
                 "Bootstrapper failed to generate any demonstrations. "
@@ -230,23 +306,36 @@ async def evaluate_one(idx: int, input_text: str, expected: Optional[str]) -> Op
         return demo_sets
 
     def _create_demonstration_sets(
-        self, 
-        bootstrapped_demonstrations: List[Demonstration], 
-        labeled_demonstrations: List[Demonstration]
+        self,
+        bootstrapped_demonstrations: List[Demonstration],
+        labeled_demonstrations: List[Demonstration],
     ) -> List[DemonstrationSet]:
-        
-        demo_sets: List[DemonstrationSet] = [DemonstrationSet(demonstrations=[], id="0-shot")]
+
+        demo_sets: List[DemonstrationSet] = [
+            DemonstrationSet(demonstrations=[], id="0-shot")
+        ]
 
         for _ in range(self.num_demonstration_sets):
             demos: List[Demonstration] = []
 
             if bootstrapped_demonstrations:
-                n_boot = min(self.max_bootstrapped_demonstrations, len(bootstrapped_demonstrations))
-                demos.extend(self.random_state.sample(bootstrapped_demonstrations, n_boot))
+                n_boot = min(
+                    self.max_bootstrapped_demonstrations,
+                    len(bootstrapped_demonstrations),
+                )
+                demos.extend(
+                    self.random_state.sample(
+                        bootstrapped_demonstrations, n_boot
+                    )
+                )
 
             if labeled_demonstrations:
-                n_labeled = min(self.max_labeled_demonstrations, len(labeled_demonstrations))
-                labeled_sample = self.random_state.sample(labeled_demonstrations, n_labeled)
+                n_labeled = min(
+                    self.max_labeled_demonstrations, len(labeled_demonstrations)
+                )
+                labeled_sample = self.random_state.sample(
+                    labeled_demonstrations, n_labeled
+                )
                 existing_indices = {d.golden_index for d in demos}
                 for demo in labeled_sample:
                     if demo.golden_index not in existing_indices:
@@ -261,9 +350,9 @@ def _create_demonstration_sets(
 
 
 def render_prompt_with_demonstrations(
-    prompt: Prompt, 
-    demonstration_set: Optional[DemonstrationSet], 
-    max_demonstrations: int = 8
+    prompt: Prompt,
+    demonstration_set: Optional[DemonstrationSet],
+    max_demonstrations: int = 8,
 ) -> Prompt:
     from deepeval.prompt.api import PromptType, PromptMessage
 
@@ -277,14 +366,20 @@ def render_prompt_with_demonstrations(
         demo_added = False
         for msg in prompt.messages_template:
             if not demo_added and msg.role == "system":
-                new_messages.append(PromptMessage(role=msg.role, content=f"{msg.content}\n\n{demo_text}"))
+                new_messages.append(
+                    PromptMessage(
+                        role=msg.role, content=f"{msg.content}\n\n{demo_text}"
+                    )
+                )
                 demo_added = True
             else:
                 new_messages.append(msg)
 
         if not demo_added and new_messages:
             first = new_messages[0]
-            new_messages[0] = PromptMessage(role=first.role, content=f"{demo_text}\n\n{first.content}")
+            new_messages[0] = PromptMessage(
+                role=first.role, content=f"{demo_text}\n\n{first.content}"
+            )
         return Prompt(messages_template=new_messages)
     else:
-        return Prompt(text_template=f"{demo_text}\n\n{prompt.text_template}")
+        return Prompt(text_template=f"{demo_text}\n\n{prompt.text_template}")