Merge pull request stanfordnlp#585 from thomasahle/main

A few optimizer improvements

dspy/signatures/signature.py

@@ -35,7 +35,9 @@ def __call__(cls, *args, **kwargs):  # noqa: ANN002
     def __new__(mcs, signature_name, bases, namespace, **kwargs):  # noqa: N804
         # Set `str` as the default type for all fields
         raw_annotations = namespace.get("__annotations__", {})
-        for name, _field in namespace.items():
+        for name, field in namespace.items():
+            if not isinstance(field, FieldInfo):
+                continue  # Don't add types to non-field attributes
             if not name.startswith("__") and name not in raw_annotations:
                 raw_annotations[name] = str
         namespace["__annotations__"] = raw_annotations
@@ -272,9 +274,8 @@ def make_signature(
 
 
 def _parse_signature(signature: str) -> Tuple[Type, Field]:
-    pattern = r"^\s*[\w\s,:]+\s*->\s*[\w\s,:]+\s*$"
-    if not re.match(pattern, signature):
-        raise ValueError(f"Invalid signature format: '{signature}'")
+    if signature.count("->") != 1:
+        raise ValueError(f"Invalid signature format: '{signature}', must contain exactly one '->'.")
 
     fields = {}
     inputs_str, outputs_str = map(str.strip, signature.split("->"))

dspy/teleprompt/signature_opt_typed.py

@@ -1,4 +1,5 @@
 import textwrap
+from dataclasses import dataclass
 from typing import Generic, Literal, TypeVar
 
 import pydantic
@@ -99,28 +100,48 @@ class GenerateInstructionInitial(Signature, Generic[T]):
     return GenerateInstructionInitial
 
 
-class GenerateSignature(dspy.Signature, Generic[T]):
-    __doc__ = textwrap.dedent("""\
-    You are an instruction optimizer for large language models.
+def generate_with_avoidance(signatures_to_avoid: list[BaseModel]) -> type[Signature]:
+    class GenerateSignature(dspy.Signature, Generic[T]):
+        __doc__ = textwrap.dedent("""\
+        You are an instruction optimizer for large language models.
+
+        I will give some task instructions I've tried, along with their corresponding validation scores.
+        - The instructions are arranged in order based on their scores, where higher scores indicate better quality.
+        - Your task is to propose a new instruction that will lead a good language model to perform the task even better.
+        - Be creative, and think out of the box.
+        - Don't repeat instructions, descriptions and prefixes that have already been attempted.
+        """)
+
+        analysis: str = OutputField(desc="Consider what made the previous instructions good or bad.")
+        proposed_signature: T = OutputField(desc="A signature that will likely lead to a high score.")
+        score: float = OutputField(
+            desc="The expected score for the new signature. Don't write anything after this number.",
+        )
+
+        @pydantic.field_validator("proposed_signature")
+        @classmethod
+        def check_signature_not_attempted(cls, s: T) -> T:
+            if s in signatures_to_avoid:
+                raise ValueError("Never propose a signature already in the list above.")
+            return s
 
-    I will give some task instructions I've tried, along with their corresponding validation scores.
-    - The instructions are arranged in order based on their scores, where higher scores indicate better quality.
-    - Your task is to propose a new instruction that will lead a good language model to perform the task even better.
-    - Be creative, and think out of the box.
-    - Don't repeat instructions, descriptions and prefixes that have already been attempted.
-    """)
+    return GenerateSignature
 
-    analysis: str = OutputField(desc="Consider what made the previous instructions good or bad.")
-    proposed_signature: T = OutputField(desc="A signature that will likely lead to a high score.")
-    score: float = OutputField(desc="The expected score for the new signature. Don't write anything after this number.")
+
+@dataclass
+class OptimizerResult:
+    program: dspy.Program
+    signatures: list[dict[str, Signature]]
+    scores: list[float]
 
 
 def optimize_signature(
     student,
     evaluator,
     n_iterations=10,
-    strategy: Literal["best", "last"] = "best",
-    sorted_order: Literal["increasing", "decreasing"] = "increasing",
+    sorted_order: Literal["increasing", "decreasing", "chronological"] = "increasing",
+    strategy: Literal["last", "best"] = "best",
+    max_examples=20,
     # Formerly part of the constructor
     prompt_model=None,
     initial_prompts=2,
@@ -139,10 +160,12 @@ def optimize_signature(
         The evaluator to use to score the program.
     n_iterations : int, optional
         The number of iterations to run, by default 10
-    strategy : Literal["best", "last"], optional
-        The strategy to use to select the final program, by default "best"
-    sorted_order : Literal["increasing", "decreasing"], optional
+    max_examples : int, optional
+        The maximum number of examples to use for the evaluator, by default 20
+    sorted_order : Literal["increasing", "decreasing", "chronological"], optional
         The order in which to sort the scores, by default "increasing"
+    strategy : Literal["last", "best"], optional
+        The strategy to use to select the final program, by default "best"
     prompt_model : dspy.LanguageModel, optional
         The language model to use to generate prompts, by default None
     initial_prompts : int, optional
@@ -222,16 +245,22 @@ def optimize_signature(
             # TODO: Parallelize this
             for name, _p in named_predictors:
                 SignatureInfo = type(candidates[name][0])  # noqa: N806
-                generator = TypedPredictor(GenerateSignature[SignatureInfo])
-
-                demos = [
-                    dspy.Example(
-                        proposed_signature=info,
-                        score=sc,
-                    )
-                    for info, sc in zip(candidates[name], scores)
-                ]
-                demos.sort(key=(lambda x: x.score), reverse=(sorted_order == "decreasing"))
+
+                demos = [dspy.Example(proposed_signature=info, score=sc) for info, sc in zip(candidates[name], scores)]
+                if sorted_order == "chronological":
+                    demos = demos[-max_examples:]
+                elif sorted_order == "increasing":
+                    demos.sort(key=(lambda x: x.score), reverse=False)
+                    demos = demos[-max_examples:]
+                elif sorted_order == "decreasing":
+                    demos.sort(key=(lambda x: x.score), reverse=True)
+                    demos = demos[:max_examples]
+                else:
+                    raise ValueError(f"Invalid sorted_order: {sorted_order}")
+
+                # We can only tell the LM to avoid the signatures we are actually giving it as demos.
+                avoid = [ex.proposed_signature for ex in demos]
+                generator = TypedPredictor(generate_with_avoidance(avoid)[SignatureInfo])
                 generator.predictor.demos = demos
 
                 if verbose:
@@ -240,12 +269,16 @@ def optimize_signature(
                 candidates[name].append(new_signature)
 
     if strategy == "last":
-        return module
-
-    if strategy == "best":
+        pass
+    elif strategy == "best":
         i = scores.index(max(scores))
         for name, p in named_predictors:
             p.signature = candidates[name][i].to_signature()
-        return module
+    else:
+        raise ValueError(f"Invalid strategy: {strategy}")
 
-    raise ValueError(f"Invalid strategy: {strategy}")
+    return OptimizerResult(
+        program=module,
+        signatures=[{name: sigs[i].to_signature()} for name, sigs in candidates.items() for i in range(n_iterations)],
+        scores=scores,
+    )