Skip to content

auto

AutoPrincipleGenerator

Bases: BaseModel

Main class for generating and clustering evaluation principles.

Attributes:

Name Type Description
llm BaseLLM

Language model client for generating responses. Must be provided as no default value is available (default=...).
scenario str

Description of the task context or scenario. Must be provided (default=...).
generate_number int

Number of principles to generate per sample. Default is 10.
cluster_number int

Number of principles to include in the final clustered output. Default is 1.
max_retries int

Maximum number of retry attempts for generation steps. Default is 3.
generate_template Type[BaseGeneratorTemplate]

Template class used for generating principles. Default is PrincipleGenerateTemplate.
cluster_template Type[BaseGeneratorTemplate]

Template class used for clustering principles. Default is PrincipleClusterTemplate.
Source code in rm_gallery/core/reward/principle/auto.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
class AutoPrincipleGenerator(BaseModel):
    """Main class for generating and clustering evaluation principles.

    Attributes:
        llm (BaseLLM): Language model client for generating responses. Must be provided
                      as no default value is available (default=...).
        scenario (str): Description of the task context or scenario. Must be provided
                       (default=...).
        generate_number (int): Number of principles to generate per sample. Default is 10.
        cluster_number (int): Number of principles to include in the final clustered output.
                           Default is 1.
        max_retries (int): Maximum number of retry attempts for generation steps. Default is 3.
        generate_template (Type[BaseGeneratorTemplate]): Template class used for generating
                           principles. Default is PrincipleGenerateTemplate.
        cluster_template (Type[BaseGeneratorTemplate]): Template class used for clustering
                           principles. Default is PrincipleClusterTemplate.
    """

    llm: BaseLLM = Field(default=..., description="llm client")
    scenario: str = Field(default=..., description="assistant scenario")
    generate_number: int = Field(
        default=10, description="number of generated principles"
    )
    cluster_number: int = Field(default=1, description="number of clustered principles")
    max_retries: int = Field(default=3, description="max retries")
    generate_template: Type[BaseGeneratorTemplate] = Field(
        default=PrincipleGenerateTemplate,
        description="template for generating principles",
    )
    cluster_template: Type[BaseGeneratorTemplate] = Field(
        default=PrincipleClusterTemplate,
        description="template for clustering principles",
    )

    def generate(self, sample: DataSample):
        """Generate principles for a single data sample.

        Args:
            sample: Input data sample containing instruction and completions

        Returns:
            Modified sample with generated principles in metadata
        """
        # Deep copy to avoid modifying original sample
        sample = copy.deepcopy(sample)
        instruction: str = format_messages(sample.input)

        # Process completions and identify best one
        completions = [
            (output.answer.label["preference"], output.answer.content)
            for output in sample.output
        ]
        random.shuffle(completions)
        for i, (label, completion) in enumerate(completions):
            if label == "chosen":
                best = i + 1
        completions = [completion for _, completion in completions]

        # Generate prompt and get LLM response
        prompt = self.generate_template.format(
            instruction=instruction,
            completions=completions,
            preference=best,
            enable_thinking=self.llm.enable_thinking,
            scenario=self.scenario,
            number=self.generate_number,
        )

        @retry(tries=self.max_retries, delay=1.0)
        def call():
            logger.info(f"prompt: {prompt}")
            response = self.llm.simple_chat(
                prompt,
                sys_prompt="You are a professional assistant skilled in extracting key insights and summarizing information.",
            )
            result = self.generate_template.parse(response)
            sample.input[-1].additional_kwargs["generate"] = result.model_dump()
            return sample

        try:
            sample = call()
        except Exception as e:
            logger.error(f"API call failed: {str(e)}")
        return sample

    def cluster(self, samples: List[DataSample]):
        """Cluster principles across multiple samples.

        Args:
            samples: List of data samples with generated principles

        Returns:
            Dictionary of clustered principles
        """
        # Build example strings from sample principles
        examples = []
        principles = {}
        for i, sample in enumerate(samples):
            sample_principles = []
            if "generate" not in sample.input[-1].additional_kwargs:
                continue

            for key, value in (
                sample.input[-1].additional_kwargs["generate"]["principles"].items()
            ):
                sample_principles.append(f"{key}: {value}")
                principles[key] = value
            str_principles = "\n".join(sample_principles)
            str_principles = (
                f"<principles_{i+1}>\n{str_principles}\n</principles_{i+1}>"
            )
            str_instruction = f"<instruction_{i+1}>\n{format_messages(sample.input)}\n</instruction_{i+1}>"
            examples.append(
                f"<example_{i+1}>\n{str_instruction}\n{str_principles}\n</example_{i+1}>\n\n"
            )

        str_examples = "\n".join(examples)
        logger.info("===RAW EXAMPLES===\n" + str_examples)

        # Get clustered principles from LLM
        @retry(tries=self.max_retries, delay=1.0)
        def call():
            response = self.llm.simple_chat(
                self.cluster_template.format(
                    scenario=self.scenario,
                    examples=str_examples,
                    enable_thinking=self.llm.enable_thinking,
                    number=self.cluster_number,
                ),
                sys_prompt="You are a skilled professional assistant focusing on induction and summarization.",
            )
            result = self.cluster_template.parse(response)
            logger.info("===CLUSTER RESULT===\n" + result.model_dump_json())
            return result.principles

        try:
            principles = call()
        except Exception as e:
            principles = {}
            logger.error(f"API call failed: {str(e)}")
        return principles

    def run_batch(
        self, samples: List[DataSample], thread_pool: ThreadPoolExecutor
    ) -> Dict[str, str]:
        """Process multiple samples in parallel.

        Args:
            samples: List of input data samples
            thread_pool: Executor for parallel processing

        Returns:
            Dictionary of clustered principles from all samples
        """
        # Submit generation tasks to thread pool
        futures = [thread_pool.submit(self.generate, sample) for sample in samples]
        wait(futures, return_when=ALL_COMPLETED)
        samples = [future.result() for future in futures]

        # Cluster results across all generated samples
        return self.cluster(samples)

cluster(samples)

Cluster principles across multiple samples.

Parameters:

Name Type Description Default
samples List[DataSample]

List of data samples with generated principles

 required

Returns:

Type Description

Dictionary of clustered principles
Source code in rm_gallery/core/reward/principle/auto.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
def cluster(self, samples: List[DataSample]):
    """Cluster principles across multiple samples.

    Args:
        samples: List of data samples with generated principles

    Returns:
        Dictionary of clustered principles
    """
    # Build example strings from sample principles
    examples = []
    principles = {}
    for i, sample in enumerate(samples):
        sample_principles = []
        if "generate" not in sample.input[-1].additional_kwargs:
            continue

        for key, value in (
            sample.input[-1].additional_kwargs["generate"]["principles"].items()
        ):
            sample_principles.append(f"{key}: {value}")
            principles[key] = value
        str_principles = "\n".join(sample_principles)
        str_principles = (
            f"<principles_{i+1}>\n{str_principles}\n</principles_{i+1}>"
        )
        str_instruction = f"<instruction_{i+1}>\n{format_messages(sample.input)}\n</instruction_{i+1}>"
        examples.append(
            f"<example_{i+1}>\n{str_instruction}\n{str_principles}\n</example_{i+1}>\n\n"
        )

    str_examples = "\n".join(examples)
    logger.info("===RAW EXAMPLES===\n" + str_examples)

    # Get clustered principles from LLM
    @retry(tries=self.max_retries, delay=1.0)
    def call():
        response = self.llm.simple_chat(
            self.cluster_template.format(
                scenario=self.scenario,
                examples=str_examples,
                enable_thinking=self.llm.enable_thinking,
                number=self.cluster_number,
            ),
            sys_prompt="You are a skilled professional assistant focusing on induction and summarization.",
        )
        result = self.cluster_template.parse(response)
        logger.info("===CLUSTER RESULT===\n" + result.model_dump_json())
        return result.principles

    try:
        principles = call()
    except Exception as e:
        principles = {}
        logger.error(f"API call failed: {str(e)}")
    return principles

generate(sample)

Generate principles for a single data sample.

Parameters:

Name Type Description Default
sample DataSample

Input data sample containing instruction and completions

 required

Returns:

Type Description

Modified sample with generated principles in metadata
Source code in rm_gallery/core/reward/principle/auto.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def generate(self, sample: DataSample):
    """Generate principles for a single data sample.

    Args:
        sample: Input data sample containing instruction and completions

    Returns:
        Modified sample with generated principles in metadata
    """
    # Deep copy to avoid modifying original sample
    sample = copy.deepcopy(sample)
    instruction: str = format_messages(sample.input)

    # Process completions and identify best one
    completions = [
        (output.answer.label["preference"], output.answer.content)
        for output in sample.output
    ]
    random.shuffle(completions)
    for i, (label, completion) in enumerate(completions):
        if label == "chosen":
            best = i + 1
    completions = [completion for _, completion in completions]

    # Generate prompt and get LLM response
    prompt = self.generate_template.format(
        instruction=instruction,
        completions=completions,
        preference=best,
        enable_thinking=self.llm.enable_thinking,
        scenario=self.scenario,
        number=self.generate_number,
    )

    @retry(tries=self.max_retries, delay=1.0)
    def call():
        logger.info(f"prompt: {prompt}")
        response = self.llm.simple_chat(
            prompt,
            sys_prompt="You are a professional assistant skilled in extracting key insights and summarizing information.",
        )
        result = self.generate_template.parse(response)
        sample.input[-1].additional_kwargs["generate"] = result.model_dump()
        return sample

    try:
        sample = call()
    except Exception as e:
        logger.error(f"API call failed: {str(e)}")
    return sample

run_batch(samples, thread_pool)

Process multiple samples in parallel.

Parameters:

Name Type Description Default
samples List[DataSample]

List of input data samples

 required
thread_pool ThreadPoolExecutor

Executor for parallel processing

 required

Returns:

Type Description
Dict[str, str]

Dictionary of clustered principles from all samples
Source code in rm_gallery/core/reward/principle/auto.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def run_batch(
    self, samples: List[DataSample], thread_pool: ThreadPoolExecutor
) -> Dict[str, str]:
    """Process multiple samples in parallel.

    Args:
        samples: List of input data samples
        thread_pool: Executor for parallel processing

    Returns:
        Dictionary of clustered principles from all samples
    """
    # Submit generation tasks to thread pool
    futures = [thread_pool.submit(self.generate, sample) for sample in samples]
    wait(futures, return_when=ALL_COMPLETED)
    samples = [future.result() for future in futures]

    # Cluster results across all generated samples
    return self.cluster(samples)

BaseGeneratorTemplate

Bases: BasePromptTemplate

Base template class for principle generation tasks.

Attributes:

Name Type Description
principles Dict[str, str]

Dictionary mapping principle phrases to descriptions
Source code in rm_gallery/core/reward/principle/auto.py 
18
19
20
21
22
23
24
25
26
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
class BaseGeneratorTemplate(BasePromptTemplate):
    """Base template class for principle generation tasks.

    Attributes:
        principles: Dictionary mapping principle phrases to descriptions
    """

    principles: Dict[str, str] = Field(
        default=...,
        description="""```json
{
    "{phrase}": "{description}",
    ...
}
```""",
    )

    @classmethod
    def parse(cls, text: str):
        """Parse response text into structured principles dictionary.

        Args:
            text: Raw response text containing JSON-formatted principles

        Returns:
            cls instance with parsed principles
        """
        contents = cls._parse(text)

        json_pattern = r"```json(.*?)```"
        json_dict = re.findall(json_pattern, contents["principles"], re.DOTALL)
        json_dict = json_dict[0] if len(json_dict) > 0 else "{}"

        try:
            parsed_dict = json.loads(json_dict)
        except json.JSONDecodeError:
            pattern = r'"(.*?)"\s*:\s*"(.*?)"'
            matches = re.findall(pattern, json_dict)
            parsed_dict = {key: value for key, value in matches}

        return cls(
            think=contents["think"],
            principles=parsed_dict,
        )

parse(text) classmethod

Parse response text into structured principles dictionary.

Parameters:

Name Type Description Default
text str

Raw response text containing JSON-formatted principles

 required

Returns:

Type Description

cls instance with parsed principles
Source code in rm_gallery/core/reward/principle/auto.py 
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
@classmethod
def parse(cls, text: str):
    """Parse response text into structured principles dictionary.

    Args:
        text: Raw response text containing JSON-formatted principles

    Returns:
        cls instance with parsed principles
    """
    contents = cls._parse(text)

    json_pattern = r"```json(.*?)```"
    json_dict = re.findall(json_pattern, contents["principles"], re.DOTALL)
    json_dict = json_dict[0] if len(json_dict) > 0 else "{}"

    try:
        parsed_dict = json.loads(json_dict)
    except json.JSONDecodeError:
        pattern = r'"(.*?)"\s*:\s*"(.*?)"'
        matches = re.findall(pattern, json_dict)
        parsed_dict = {key: value for key, value in matches}

    return cls(
        think=contents["think"],
        principles=parsed_dict,
    )

PrincipleClusterTemplate

Bases: BaseGeneratorTemplate

Template for clustering and summarizing generated principles.

Source code in rm_gallery/core/reward/principle/auto.py 
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
class PrincipleClusterTemplate(BaseGeneratorTemplate):
    """Template for clustering and summarizing generated principles."""

    @classmethod
    def format(cls, examples: str, scenario: str, number: int, **kwargs) -> str:
        """Format prompt for principle clustering task.

        Args:
            examples: XML-formatted example principles
            scenario: Task context description
            number: Maximum number of clustered principles
            **kwargs: Additional template parameters

        Returns:
            Formatted prompt string
        """
        return f"""## Overview
You will be provided with a set of examples with instruction and pre-generated principles in the scenario.
Please summarize some general principles from the examples that can help another assistant to determine which one completion is superior to the others in the scenario.

## Requirements for Principles
(1) Principles are presented from most important to least important.
(2) Principles should be as critical as possible.
(3) Each principle should consist of a brief phrase accompanied by a single sentence description.
(4) The number of principles should be LESS THAN OR EQUAL TO {number}.
(5) Focus on summarizing recurring candidate principles.

## Input
### Scenario
{scenario}

### Examples
{examples}

## Output Format Requirements
{cls.schema(**kwargs)}
"""

format(examples, scenario, number, **kwargs) classmethod

Format prompt for principle clustering task.

Parameters:

Name Type Description Default
examples str

XML-formatted example principles

 required
scenario str

Task context description

 required
number int

Maximum number of clustered principles

 required
**kwargs

Additional template parameters

 {}

Returns:

Type Description
str

Formatted prompt string
Source code in rm_gallery/core/reward/principle/auto.py 
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
    @classmethod
    def format(cls, examples: str, scenario: str, number: int, **kwargs) -> str:
        """Format prompt for principle clustering task.

        Args:
            examples: XML-formatted example principles
            scenario: Task context description
            number: Maximum number of clustered principles
            **kwargs: Additional template parameters

        Returns:
            Formatted prompt string
        """
        return f"""## Overview
You will be provided with a set of examples with instruction and pre-generated principles in the scenario.
Please summarize some general principles from the examples that can help another assistant to determine which one completion is superior to the others in the scenario.

## Requirements for Principles
(1) Principles are presented from most important to least important.
(2) Principles should be as critical as possible.
(3) Each principle should consist of a brief phrase accompanied by a single sentence description.
(4) The number of principles should be LESS THAN OR EQUAL TO {number}.
(5) Focus on summarizing recurring candidate principles.

## Input
### Scenario
{scenario}

### Examples
{examples}

## Output Format Requirements
{cls.schema(**kwargs)}
"""

PrincipleGenerateTemplate

Bases: BaseGeneratorTemplate

Template for generating evaluation principles from completion comparisons.

Source code in rm_gallery/core/reward/principle/auto.py 
 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
class PrincipleGenerateTemplate(BaseGeneratorTemplate):
    """Template for generating evaluation principles from completion comparisons."""

    @classmethod
    def format(
        cls,
        scenario: str,
        instruction: str,
        completions: List[str],
        preference: str | int,
        number: int,
        **kwargs,
    ) -> str:
        """Format prompt for principle generation task.

        Args:
            scenario: Task context/scenario description
            instruction: Original instruction text
            completions: List of completion texts to compare
            preference: Index/ID of preferred completion
            number: Maximum number of principles to generate
            **kwargs: Additional template parameters

        Returns:
            Formatted prompt string
        """
        completion_str = ""
        for i, completion in enumerate(completions):
            completion_str += (
                f"<completion_{i + 1}>\n{completion}\n</completion_{i + 1}>\n\n"
            )

        return f"""## Overview
You will be provided with an example of instruction and completions in a task scenario.
Please propose some general principles from the scenario that can help another assistant to determine which one completion is superior to the others in the scenario.

## Requirements for Principles
(1) Principles target some general standards of the "scenario".
(2) Principles are presented from most important to least important.
(3) Principles should be as critical as possible.
(4) Each principle should consist of a brief phrase accompanied by a single sentence description.
(5) The number of principles should be LESS THAN OR EQUAL TO {number}.

## Input
### Scenario
{scenario}

### Instruction
{instruction}

### Completions
{completion_str}

### Preference
Completion {preference} is the best.

## Output Format Requirements
{cls.schema(**kwargs)}
"""

format(scenario, instruction, completions, preference, number, **kwargs) classmethod

Format prompt for principle generation task.

Parameters:

Name Type Description Default
scenario str

Task context/scenario description

 required
instruction str

Original instruction text

 required
completions List[str]

List of completion texts to compare

 required
preference str | int

Index/ID of preferred completion

 required
number int

Maximum number of principles to generate

 required
**kwargs

Additional template parameters

 {}

Returns:

Type Description
str

Formatted prompt string
Source code in rm_gallery/core/reward/principle/auto.py 
 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
    @classmethod
    def format(
        cls,
        scenario: str,
        instruction: str,
        completions: List[str],
        preference: str | int,
        number: int,
        **kwargs,
    ) -> str:
        """Format prompt for principle generation task.

        Args:
            scenario: Task context/scenario description
            instruction: Original instruction text
            completions: List of completion texts to compare
            preference: Index/ID of preferred completion
            number: Maximum number of principles to generate
            **kwargs: Additional template parameters

        Returns:
            Formatted prompt string
        """
        completion_str = ""
        for i, completion in enumerate(completions):
            completion_str += (
                f"<completion_{i + 1}>\n{completion}\n</completion_{i + 1}>\n\n"
            )

        return f"""## Overview
You will be provided with an example of instruction and completions in a task scenario.
Please propose some general principles from the scenario that can help another assistant to determine which one completion is superior to the others in the scenario.

## Requirements for Principles
(1) Principles target some general standards of the "scenario".
(2) Principles are presented from most important to least important.
(3) Principles should be as critical as possible.
(4) Each principle should consist of a brief phrase accompanied by a single sentence description.
(5) The number of principles should be LESS THAN OR EQUAL TO {number}.

## Input
### Scenario
{scenario}

### Instruction
{instruction}

### Completions
{completion_str}

### Preference
Completion {preference} is the best.

## Output Format Requirements
{cls.schema(**kwargs)}
"""