Skip to content

OPRO

OPRO

Bases: Optimizer

Optimization by PROmpting (OPRO) optimizer

Use a language model to optimize the prompt's instructions.

Example:

import synalinks
import asyncio

async def main():
    # ... your program definition

    program.compile(
        reward=synalinks.rewards.ExactMatch(),
        optimizer=synalinks.optimizers.OPRO(
            language_model=language_model, # The language model to use
            k_best=10, # The number of best examples/instructions to provide to the LM
        ),
    )

    history = await program.fit(...)
References

Parameters:

Name Type Description Default
language_model LanguageModel

The language model to use.

 None
k_best int

The max number of best predictions and instructions to provide to the optimizer (default 10).

 10
Source code in synalinks/src/optimizers/opro.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
@synalinks_export("synalinks.optimizers.OPRO")
class OPRO(Optimizer):
    """Optimization by PROmpting (OPRO) optimizer

    Use a language model to optimize the prompt's instructions.

    Example:

    ```python
    import synalinks
    import asyncio

    async def main():
        # ... your program definition

        program.compile(
            reward=synalinks.rewards.ExactMatch(),
            optimizer=synalinks.optimizers.OPRO(
                language_model=language_model, # The language model to use
                k_best=10, # The number of best examples/instructions to provide to the LM
            ),
        )

        history = await program.fit(...)
    ```

    References:
        - [Large Language Models as Optimizers](https://arxiv.org/abs/2309.03409)

    Args:
        language_model (LanguageModel): The language model to use.
        k_best (int): The max number of best predictions and instructions
            to provide to the optimizer (default 10).
    """

    def __init__(
        self,
        language_model=None,
        k_best=10,
        name=None,
        description=None,
        **kwargs,
    ):
        super().__init__(
            name=name,
            description=description,
            data_model=OPROOptimizedVariable,
        )
        self.language_model = language_model
        self.k_best = k_best

    async def build(self, variables):
        opro_inputs = Input(data_model=OPROInputs)
        opro_outputs = await Generator(
            language_model=self.language_model,
            data_model=Instructions,
            instructions=[
                "Your task is to generate instructions that maximize rewards.",
                "Below are some previous instructions with their reward.",
                "Generate instructions that is different from all the instructions.",
                (
                    "The instructions should be concise, effective and generally"
                    " applicable to all predictions below."
                ),
            ],
        )(opro_inputs)

        self.opro = Program(
            inputs=opro_inputs,
            outputs=opro_outputs,
            name="opro",
            description="OPRO Program",
        )
        self.built = True

    async def optimize(self, trainable_variable, reward=None):
        """Perform a backprop/optimization on a single variable."""
        # Backpropagate predictions reward
        predictions = trainable_variable.get("predictions")
        backpropagated_predictions = []
        backprop_pred_nb = 0
        for p in predictions:
            if p["reward"] is None:
                p["reward"] = reward
                backprop_pred_nb += 1
            backpropagated_predictions.append(p)
        if backprop_pred_nb > 0:
            trainable_variable.update({"predictions": backpropagated_predictions})
            # Backpropagate instructions reward
            instructions_predictions = trainable_variable.get("instructions_predictions")
            instructions = trainable_variable.get("instructions")
            instructions.update({"reward": reward})
            instructions_predictions.append(instructions)
            # Get the k best predictions (sorted by reward)
            sorted_predictions = sorted(
                backpropagated_predictions,
                key=lambda x: x["reward"] if x["reward"] is not None else float("-inf"),
                reverse=True,
            )
            top_k_predictions = sorted_predictions[: self.k_best]
            # Get the k best instructions (sorted by reward)
            sorted_instructions = sorted(
                trainable_variable.get("instructions_predictions"),
                key=lambda x: x["reward"] if x["reward"] is not None else float("-inf"),
                reverse=True,
            )
            top_k_instructions = sorted_instructions[: self.k_best]
            # Prepare inputs for OPRO
            inputs = OPROInputs(
                predictions=top_k_predictions,
                instructions_predictions=top_k_instructions,
            )
            new_instructions = await self.opro(inputs)
            trainable_variable.update({"instructions": new_instructions.get_json()})

    async def finalize(self, trainable_variable):
        """Finalize the optimization of a single variable (cleanup/scaling etc.)."""
        trainable_variable.update({"instructions_predictions": []})

    def get_config(self):
        config = {
            "k_best": self.k_best,
            "name": self.name,
            "description": self.description,
        }
        language_model_config = {
            "language_model": serialization_lib.serialize_synalinks_object(
                self.language_model,
            )
        }
        return {**config, **language_model_config}

    @classmethod
    def from_config(cls, config):
        language_model = serialization_lib.deserialize_synalinks_object(
            config.pop("language_model"),
        )
        return cls(language_model=language_model, **config)

finalize(trainable_variable) async

Finalize the optimization of a single variable (cleanup/scaling etc.).

Source code in synalinks/src/optimizers/opro.py 
142
143
144
async def finalize(self, trainable_variable):
    """Finalize the optimization of a single variable (cleanup/scaling etc.)."""
    trainable_variable.update({"instructions_predictions": []})

optimize(trainable_variable, reward=None) async

Perform a backprop/optimization on a single variable.

Source code in synalinks/src/optimizers/opro.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
async def optimize(self, trainable_variable, reward=None):
    """Perform a backprop/optimization on a single variable."""
    # Backpropagate predictions reward
    predictions = trainable_variable.get("predictions")
    backpropagated_predictions = []
    backprop_pred_nb = 0
    for p in predictions:
        if p["reward"] is None:
            p["reward"] = reward
            backprop_pred_nb += 1
        backpropagated_predictions.append(p)
    if backprop_pred_nb > 0:
        trainable_variable.update({"predictions": backpropagated_predictions})
        # Backpropagate instructions reward
        instructions_predictions = trainable_variable.get("instructions_predictions")
        instructions = trainable_variable.get("instructions")
        instructions.update({"reward": reward})
        instructions_predictions.append(instructions)
        # Get the k best predictions (sorted by reward)
        sorted_predictions = sorted(
            backpropagated_predictions,
            key=lambda x: x["reward"] if x["reward"] is not None else float("-inf"),
            reverse=True,
        )
        top_k_predictions = sorted_predictions[: self.k_best]
        # Get the k best instructions (sorted by reward)
        sorted_instructions = sorted(
            trainable_variable.get("instructions_predictions"),
            key=lambda x: x["reward"] if x["reward"] is not None else float("-inf"),
            reverse=True,
        )
        top_k_instructions = sorted_instructions[: self.k_best]
        # Prepare inputs for OPRO
        inputs = OPROInputs(
            predictions=top_k_predictions,
            instructions_predictions=top_k_instructions,
        )
        new_instructions = await self.opro(inputs)
        trainable_variable.update({"instructions": new_instructions.get_json()})