kiln_ai.adapters.prompt_builders
1from abc import ABCMeta, abstractmethod 2 3from kiln_ai.datamodel import PromptGenerators, PromptId, Task, TaskRun 4from kiln_ai.utils.exhaustive_error import raise_exhaustive_enum_error 5 6 7class BasePromptBuilder(metaclass=ABCMeta): 8 """Base class for building prompts from tasks. 9 10 Provides the core interface and basic functionality for prompt builders. 11 """ 12 13 def __init__(self, task: Task): 14 """Initialize the prompt builder with a task. 15 16 Args: 17 task (Task): The task containing instructions and requirements. 18 """ 19 self.task = task 20 21 def prompt_id(self) -> str | None: 22 """Returns the ID of the prompt, scoped to this builder. 23 24 Returns: 25 str | None: The ID of the prompt, or None if not set. 26 """ 27 return None 28 29 def build_prompt(self, include_json_instructions) -> str: 30 """Build and return the complete prompt string. 31 32 Returns: 33 str: The constructed prompt. 34 """ 35 prompt = self.build_base_prompt() 36 37 if include_json_instructions and self.task.output_schema(): 38 prompt = ( 39 prompt 40 + f"\n\n# Format Instructions\n\nReturn a JSON object conforming to the following schema:\n```\n{self.task.output_schema()}\n```" 41 ) 42 43 return prompt 44 45 @abstractmethod 46 def build_base_prompt(self) -> str: 47 """Build and return the complete prompt string. 48 49 Returns: 50 str: The constructed prompt. 51 """ 52 pass 53 54 def chain_of_thought_prompt(self) -> str | None: 55 """Build and return the chain of thought prompt string. 56 57 Returns: 58 str: The constructed chain of thought prompt. 59 """ 60 return None 61 62 def build_prompt_for_ui(self) -> str: 63 """Build a prompt for the UI. It includes additional instructions (like chain of thought), even if they are passed to the model in stages. 64 65 Designed for end-user consumption, not for model consumption. 66 67 Returns: 68 str: The constructed prompt string. 69 """ 70 base_prompt = self.build_prompt(include_json_instructions=False) 71 cot_prompt = self.chain_of_thought_prompt() 72 if cot_prompt: 73 base_prompt += "\n# Thinking Instructions\n\n" + cot_prompt 74 return base_prompt 75 76 77class SimplePromptBuilder(BasePromptBuilder): 78 """A basic prompt builder that combines task instruction with requirements.""" 79 80 def build_base_prompt(self) -> str: 81 """Build a simple prompt with instruction and requirements. 82 83 Returns: 84 str: The constructed prompt string. 85 """ 86 base_prompt = self.task.instruction 87 88 if len(self.task.requirements) > 0: 89 base_prompt += ( 90 "\n\nYour response should respect the following requirements:\n" 91 ) 92 # iterate requirements, formatting them in numbereed list like 1) task.instruction\n2)... 93 for i, requirement in enumerate(self.task.requirements): 94 base_prompt += f"{i + 1}) {requirement.instruction}\n" 95 96 return base_prompt 97 98 99class ShortPromptBuilder(BasePromptBuilder): 100 """A prompt builder that includes a the base prompt but excludes the requirements.""" 101 102 def build_base_prompt(self) -> str: 103 """Build a short prompt with just the base prompt, no requirements. 104 105 Returns: 106 str: The constructed prompt string. 107 """ 108 return self.task.instruction 109 110 111class MultiShotPromptBuilder(BasePromptBuilder): 112 """A prompt builder that includes multiple examples in the prompt.""" 113 114 @classmethod 115 def example_count(cls) -> int: 116 """Get the maximum number of examples to include in the prompt. 117 118 Returns: 119 int: The maximum number of examples (default 25). 120 """ 121 return 25 122 123 def build_base_prompt(self) -> str: 124 """Build a prompt with instruction, requirements, and multiple examples. 125 126 Returns: 127 str: The constructed prompt string with examples. 128 """ 129 base_prompt = f"# Instruction\n\n{self.task.instruction}\n\n" 130 131 if len(self.task.requirements) > 0: 132 base_prompt += "# Requirements\n\nYour response should respect the following requirements:\n" 133 for i, requirement in enumerate(self.task.requirements): 134 base_prompt += f"{i + 1}) {requirement.instruction}\n" 135 base_prompt += "\n" 136 137 valid_examples = self.collect_examples() 138 139 if len(valid_examples) == 0: 140 return base_prompt 141 142 base_prompt += "# Example Outputs\n\n" 143 for i, example in enumerate(valid_examples): 144 base_prompt += self.prompt_section_for_example(i, example) 145 146 return base_prompt 147 148 def prompt_section_for_example(self, index: int, example: TaskRun) -> str: 149 # Prefer repaired output if it exists, otherwise use the regular output 150 output = example.repaired_output or example.output 151 return f"## Example {index + 1}\n\nInput: {example.input}\nOutput: {output.output}\n\n" 152 153 def collect_examples(self) -> list[TaskRun]: 154 valid_examples: list[TaskRun] = [] 155 runs = self.task.runs(readonly=True) 156 157 # first pass, we look for repaired outputs. These are the best examples. 158 for run in runs: 159 if len(valid_examples) >= self.__class__.example_count(): 160 break 161 if run.repaired_output is not None: 162 valid_examples.append(run) 163 164 # second pass, we look for high quality outputs (rating based) 165 # Minimum is "high_quality" (4 star in star rating scale), then sort by rating 166 # exclude repaired outputs as they were used above 167 runs_with_rating = [ 168 run 169 for run in runs 170 if run.output.rating is not None 171 and run.output.rating.value is not None 172 and run.output.rating.is_high_quality() 173 and run.repaired_output is None 174 ] 175 runs_with_rating.sort( 176 key=lambda x: (x.output.rating and x.output.rating.value) or 0, reverse=True 177 ) 178 for run in runs_with_rating: 179 if len(valid_examples) >= self.__class__.example_count(): 180 break 181 valid_examples.append(run) 182 return valid_examples 183 184 185class FewShotPromptBuilder(MultiShotPromptBuilder): 186 """A prompt builder that includes a small number of examples in the prompt.""" 187 188 @classmethod 189 def example_count(cls) -> int: 190 """Get the maximum number of examples to include in the prompt. 191 192 Returns: 193 int: The maximum number of examples (4). 194 """ 195 return 4 196 197 198class RepairsPromptBuilder(MultiShotPromptBuilder): 199 """A prompt builder that includes multiple examples in the prompt, including repaired instructions describing what was wrong, and how it was fixed.""" 200 201 def prompt_section_for_example(self, index: int, example: TaskRun) -> str: 202 if ( 203 not example.repaired_output 204 or not example.repair_instructions 205 or not example.repaired_output.output 206 ): 207 return super().prompt_section_for_example(index, example) 208 209 prompt_section = f"## Example {index + 1}\n\nInput: {example.input}\n\n" 210 prompt_section += ( 211 f"Initial Output Which Was Insufficient: {example.output.output}\n\n" 212 ) 213 prompt_section += f"Instructions On How to Improve the Initial Output: {example.repair_instructions}\n\n" 214 prompt_section += ( 215 f"Repaired Output Which is Sufficient: {example.repaired_output.output}\n\n" 216 ) 217 return prompt_section 218 219 220def chain_of_thought_prompt(task: Task) -> str: 221 """Standard implementation to build and return the chain of thought prompt string. 222 223 Returns: 224 str: The constructed chain of thought prompt. 225 """ 226 227 cot_instruction = task.thinking_instruction 228 if not cot_instruction: 229 cot_instruction = "Think step by step, explaining your reasoning." 230 231 return cot_instruction 232 233 234class SimpleChainOfThoughtPromptBuilder(SimplePromptBuilder): 235 """A prompt builder that includes a chain of thought prompt on top of the simple prompt.""" 236 237 def chain_of_thought_prompt(self) -> str | None: 238 return chain_of_thought_prompt(self.task) 239 240 241class FewShotChainOfThoughtPromptBuilder(FewShotPromptBuilder): 242 """A prompt builder that includes a chain of thought prompt on top of the few shot prompt.""" 243 244 def chain_of_thought_prompt(self) -> str | None: 245 return chain_of_thought_prompt(self.task) 246 247 248class MultiShotChainOfThoughtPromptBuilder(MultiShotPromptBuilder): 249 """A prompt builder that includes a chain of thought prompt on top of the multi shot prompt.""" 250 251 def chain_of_thought_prompt(self) -> str | None: 252 return chain_of_thought_prompt(self.task) 253 254 255class SavedPromptBuilder(BasePromptBuilder): 256 """A prompt builder that looks up a static prompt.""" 257 258 def __init__(self, task: Task, prompt_id: str): 259 super().__init__(task) 260 prompt_model = next( 261 ( 262 prompt 263 for prompt in task.prompts(readonly=True) 264 if prompt.id == prompt_id 265 ), 266 None, 267 ) 268 if not prompt_model: 269 raise ValueError(f"Prompt ID not found: {prompt_id}") 270 self.prompt_model = prompt_model 271 272 def prompt_id(self) -> str | None: 273 return self.prompt_model.id 274 275 def build_base_prompt(self) -> str: 276 """Returns a saved prompt. 277 278 Returns: 279 str: The prompt string. 280 """ 281 return self.prompt_model.prompt 282 283 def chain_of_thought_prompt(self) -> str | None: 284 return self.prompt_model.chain_of_thought_instructions 285 286 287class TaskRunConfigPromptBuilder(BasePromptBuilder): 288 """A prompt builder that looks up a static prompt in a task run config.""" 289 290 def __init__(self, task: Task, run_config_prompt_id: str): 291 parts = run_config_prompt_id.split("::") 292 if len(parts) != 4: 293 raise ValueError( 294 f"Invalid task run config prompt ID: {run_config_prompt_id}. Expected format: 'task_run_config::[project_id]::[task_id]::[run_config_id]'." 295 ) 296 297 task_id = parts[2] 298 if task_id != task.id: 299 raise ValueError( 300 f"Task run config prompt ID: {run_config_prompt_id}. Task ID mismatch. Expected: {task.id}, got: {task_id}." 301 ) 302 303 run_config_id = parts[3] 304 run_config = next( 305 ( 306 run_config 307 for run_config in task.run_configs(readonly=True) 308 if run_config.id == run_config_id 309 ), 310 None, 311 ) 312 if not run_config: 313 raise ValueError( 314 f"Task run config ID not found: {run_config_id} for prompt id {run_config_prompt_id}" 315 ) 316 if run_config.prompt is None: 317 raise ValueError( 318 f"Task run config ID {run_config_id} does not have a stored prompt. Used as prompt id {run_config_prompt_id}" 319 ) 320 321 # Load the prompt from the model 322 self.prompt = run_config.prompt.prompt 323 self.cot_prompt = run_config.prompt.chain_of_thought_instructions 324 self.id = run_config_prompt_id 325 326 super().__init__(task) 327 328 def prompt_id(self) -> str | None: 329 return self.id 330 331 def build_base_prompt(self) -> str: 332 return self.prompt 333 334 def chain_of_thought_prompt(self) -> str | None: 335 return self.cot_prompt 336 337 338class FineTunePromptBuilder(BasePromptBuilder): 339 """A prompt builder that looks up a fine-tune prompt.""" 340 341 def __init__(self, task: Task, nested_fine_tune_id: str): 342 super().__init__(task) 343 344 # IDs are in project_id::task_id::fine_tune_id format 345 self.full_fine_tune_id = nested_fine_tune_id 346 parts = nested_fine_tune_id.split("::") 347 if len(parts) != 3: 348 raise ValueError( 349 f"Invalid fine-tune ID format. Expected 'project_id::task_id::fine_tune_id', got: {nested_fine_tune_id}" 350 ) 351 fine_tune_id = parts[2] 352 353 fine_tune_model = next( 354 ( 355 fine_tune 356 for fine_tune in task.finetunes(readonly=True) 357 if fine_tune.id == fine_tune_id 358 ), 359 None, 360 ) 361 if not fine_tune_model: 362 raise ValueError(f"Fine-tune ID not found: {fine_tune_id}") 363 self.fine_tune_model = fine_tune_model 364 365 def prompt_id(self) -> str | None: 366 return self.full_fine_tune_id 367 368 def build_base_prompt(self) -> str: 369 return self.fine_tune_model.system_message 370 371 def chain_of_thought_prompt(self) -> str | None: 372 return self.fine_tune_model.thinking_instructions 373 374 375# Our UI has some names that are not the same as the class names, which also hint parameters. 376def prompt_builder_from_id(prompt_id: PromptId, task: Task) -> BasePromptBuilder: 377 """Convert a name used in the UI to the corresponding prompt builder class. 378 379 Args: 380 prompt_id (PromptId): The prompt ID. 381 382 Returns: 383 type[BasePromptBuilder]: The corresponding prompt builder class. 384 385 Raises: 386 ValueError: If the UI name is not recognized. 387 """ 388 389 # Saved prompts are prefixed with "id::" 390 if prompt_id.startswith("id::"): 391 prompt_id = prompt_id[4:] 392 return SavedPromptBuilder(task, prompt_id) 393 394 # Task run config prompts are prefixed with "task_run_config::" 395 # task_run_config::[project_id]::[task_id]::[run_config_id] 396 if prompt_id.startswith("task_run_config::"): 397 return TaskRunConfigPromptBuilder(task, prompt_id) 398 399 # Fine-tune prompts are prefixed with "fine_tune_prompt::" 400 if prompt_id.startswith("fine_tune_prompt::"): 401 prompt_id = prompt_id[18:] 402 return FineTunePromptBuilder(task, prompt_id) 403 404 # Check if the prompt_id matches any enum value 405 if prompt_id not in [member.value for member in PromptGenerators]: 406 raise ValueError(f"Unknown prompt generator: {prompt_id}") 407 typed_prompt_generator = PromptGenerators(prompt_id) 408 409 match typed_prompt_generator: 410 case PromptGenerators.SIMPLE: 411 return SimplePromptBuilder(task) 412 case PromptGenerators.SHORT: 413 return ShortPromptBuilder(task) 414 case PromptGenerators.FEW_SHOT: 415 return FewShotPromptBuilder(task) 416 case PromptGenerators.MULTI_SHOT: 417 return MultiShotPromptBuilder(task) 418 case PromptGenerators.REPAIRS: 419 return RepairsPromptBuilder(task) 420 case PromptGenerators.SIMPLE_CHAIN_OF_THOUGHT: 421 return SimpleChainOfThoughtPromptBuilder(task) 422 case PromptGenerators.FEW_SHOT_CHAIN_OF_THOUGHT: 423 return FewShotChainOfThoughtPromptBuilder(task) 424 case PromptGenerators.MULTI_SHOT_CHAIN_OF_THOUGHT: 425 return MultiShotChainOfThoughtPromptBuilder(task) 426 case _: 427 # Type checking will find missing cases 428 raise_exhaustive_enum_error(typed_prompt_generator)
8class BasePromptBuilder(metaclass=ABCMeta): 9 """Base class for building prompts from tasks. 10 11 Provides the core interface and basic functionality for prompt builders. 12 """ 13 14 def __init__(self, task: Task): 15 """Initialize the prompt builder with a task. 16 17 Args: 18 task (Task): The task containing instructions and requirements. 19 """ 20 self.task = task 21 22 def prompt_id(self) -> str | None: 23 """Returns the ID of the prompt, scoped to this builder. 24 25 Returns: 26 str | None: The ID of the prompt, or None if not set. 27 """ 28 return None 29 30 def build_prompt(self, include_json_instructions) -> str: 31 """Build and return the complete prompt string. 32 33 Returns: 34 str: The constructed prompt. 35 """ 36 prompt = self.build_base_prompt() 37 38 if include_json_instructions and self.task.output_schema(): 39 prompt = ( 40 prompt 41 + f"\n\n# Format Instructions\n\nReturn a JSON object conforming to the following schema:\n```\n{self.task.output_schema()}\n```" 42 ) 43 44 return prompt 45 46 @abstractmethod 47 def build_base_prompt(self) -> str: 48 """Build and return the complete prompt string. 49 50 Returns: 51 str: The constructed prompt. 52 """ 53 pass 54 55 def chain_of_thought_prompt(self) -> str | None: 56 """Build and return the chain of thought prompt string. 57 58 Returns: 59 str: The constructed chain of thought prompt. 60 """ 61 return None 62 63 def build_prompt_for_ui(self) -> str: 64 """Build a prompt for the UI. It includes additional instructions (like chain of thought), even if they are passed to the model in stages. 65 66 Designed for end-user consumption, not for model consumption. 67 68 Returns: 69 str: The constructed prompt string. 70 """ 71 base_prompt = self.build_prompt(include_json_instructions=False) 72 cot_prompt = self.chain_of_thought_prompt() 73 if cot_prompt: 74 base_prompt += "\n# Thinking Instructions\n\n" + cot_prompt 75 return base_prompt
Base class for building prompts from tasks.
Provides the core interface and basic functionality for prompt builders.
14 def __init__(self, task: Task): 15 """Initialize the prompt builder with a task. 16 17 Args: 18 task (Task): The task containing instructions and requirements. 19 """ 20 self.task = task
Initialize the prompt builder with a task.
Args: task (Task): The task containing instructions and requirements.
22 def prompt_id(self) -> str | None: 23 """Returns the ID of the prompt, scoped to this builder. 24 25 Returns: 26 str | None: The ID of the prompt, or None if not set. 27 """ 28 return None
Returns the ID of the prompt, scoped to this builder.
Returns: str | None: The ID of the prompt, or None if not set.
30 def build_prompt(self, include_json_instructions) -> str: 31 """Build and return the complete prompt string. 32 33 Returns: 34 str: The constructed prompt. 35 """ 36 prompt = self.build_base_prompt() 37 38 if include_json_instructions and self.task.output_schema(): 39 prompt = ( 40 prompt 41 + f"\n\n# Format Instructions\n\nReturn a JSON object conforming to the following schema:\n```\n{self.task.output_schema()}\n```" 42 ) 43 44 return prompt
Build and return the complete prompt string.
Returns: str: The constructed prompt.
46 @abstractmethod 47 def build_base_prompt(self) -> str: 48 """Build and return the complete prompt string. 49 50 Returns: 51 str: The constructed prompt. 52 """ 53 pass
Build and return the complete prompt string.
Returns: str: The constructed prompt.
55 def chain_of_thought_prompt(self) -> str | None: 56 """Build and return the chain of thought prompt string. 57 58 Returns: 59 str: The constructed chain of thought prompt. 60 """ 61 return None
Build and return the chain of thought prompt string.
Returns: str: The constructed chain of thought prompt.
63 def build_prompt_for_ui(self) -> str: 64 """Build a prompt for the UI. It includes additional instructions (like chain of thought), even if they are passed to the model in stages. 65 66 Designed for end-user consumption, not for model consumption. 67 68 Returns: 69 str: The constructed prompt string. 70 """ 71 base_prompt = self.build_prompt(include_json_instructions=False) 72 cot_prompt = self.chain_of_thought_prompt() 73 if cot_prompt: 74 base_prompt += "\n# Thinking Instructions\n\n" + cot_prompt 75 return base_prompt
Build a prompt for the UI. It includes additional instructions (like chain of thought), even if they are passed to the model in stages.
Designed for end-user consumption, not for model consumption.
Returns: str: The constructed prompt string.
78class SimplePromptBuilder(BasePromptBuilder): 79 """A basic prompt builder that combines task instruction with requirements.""" 80 81 def build_base_prompt(self) -> str: 82 """Build a simple prompt with instruction and requirements. 83 84 Returns: 85 str: The constructed prompt string. 86 """ 87 base_prompt = self.task.instruction 88 89 if len(self.task.requirements) > 0: 90 base_prompt += ( 91 "\n\nYour response should respect the following requirements:\n" 92 ) 93 # iterate requirements, formatting them in numbereed list like 1) task.instruction\n2)... 94 for i, requirement in enumerate(self.task.requirements): 95 base_prompt += f"{i + 1}) {requirement.instruction}\n" 96 97 return base_prompt
A basic prompt builder that combines task instruction with requirements.
81 def build_base_prompt(self) -> str: 82 """Build a simple prompt with instruction and requirements. 83 84 Returns: 85 str: The constructed prompt string. 86 """ 87 base_prompt = self.task.instruction 88 89 if len(self.task.requirements) > 0: 90 base_prompt += ( 91 "\n\nYour response should respect the following requirements:\n" 92 ) 93 # iterate requirements, formatting them in numbereed list like 1) task.instruction\n2)... 94 for i, requirement in enumerate(self.task.requirements): 95 base_prompt += f"{i + 1}) {requirement.instruction}\n" 96 97 return base_prompt
Build a simple prompt with instruction and requirements.
Returns: str: The constructed prompt string.
100class ShortPromptBuilder(BasePromptBuilder): 101 """A prompt builder that includes a the base prompt but excludes the requirements.""" 102 103 def build_base_prompt(self) -> str: 104 """Build a short prompt with just the base prompt, no requirements. 105 106 Returns: 107 str: The constructed prompt string. 108 """ 109 return self.task.instruction
A prompt builder that includes a the base prompt but excludes the requirements.
103 def build_base_prompt(self) -> str: 104 """Build a short prompt with just the base prompt, no requirements. 105 106 Returns: 107 str: The constructed prompt string. 108 """ 109 return self.task.instruction
Build a short prompt with just the base prompt, no requirements.
Returns: str: The constructed prompt string.
112class MultiShotPromptBuilder(BasePromptBuilder): 113 """A prompt builder that includes multiple examples in the prompt.""" 114 115 @classmethod 116 def example_count(cls) -> int: 117 """Get the maximum number of examples to include in the prompt. 118 119 Returns: 120 int: The maximum number of examples (default 25). 121 """ 122 return 25 123 124 def build_base_prompt(self) -> str: 125 """Build a prompt with instruction, requirements, and multiple examples. 126 127 Returns: 128 str: The constructed prompt string with examples. 129 """ 130 base_prompt = f"# Instruction\n\n{self.task.instruction}\n\n" 131 132 if len(self.task.requirements) > 0: 133 base_prompt += "# Requirements\n\nYour response should respect the following requirements:\n" 134 for i, requirement in enumerate(self.task.requirements): 135 base_prompt += f"{i + 1}) {requirement.instruction}\n" 136 base_prompt += "\n" 137 138 valid_examples = self.collect_examples() 139 140 if len(valid_examples) == 0: 141 return base_prompt 142 143 base_prompt += "# Example Outputs\n\n" 144 for i, example in enumerate(valid_examples): 145 base_prompt += self.prompt_section_for_example(i, example) 146 147 return base_prompt 148 149 def prompt_section_for_example(self, index: int, example: TaskRun) -> str: 150 # Prefer repaired output if it exists, otherwise use the regular output 151 output = example.repaired_output or example.output 152 return f"## Example {index + 1}\n\nInput: {example.input}\nOutput: {output.output}\n\n" 153 154 def collect_examples(self) -> list[TaskRun]: 155 valid_examples: list[TaskRun] = [] 156 runs = self.task.runs(readonly=True) 157 158 # first pass, we look for repaired outputs. These are the best examples. 159 for run in runs: 160 if len(valid_examples) >= self.__class__.example_count(): 161 break 162 if run.repaired_output is not None: 163 valid_examples.append(run) 164 165 # second pass, we look for high quality outputs (rating based) 166 # Minimum is "high_quality" (4 star in star rating scale), then sort by rating 167 # exclude repaired outputs as they were used above 168 runs_with_rating = [ 169 run 170 for run in runs 171 if run.output.rating is not None 172 and run.output.rating.value is not None 173 and run.output.rating.is_high_quality() 174 and run.repaired_output is None 175 ] 176 runs_with_rating.sort( 177 key=lambda x: (x.output.rating and x.output.rating.value) or 0, reverse=True 178 ) 179 for run in runs_with_rating: 180 if len(valid_examples) >= self.__class__.example_count(): 181 break 182 valid_examples.append(run) 183 return valid_examples
A prompt builder that includes multiple examples in the prompt.
115 @classmethod 116 def example_count(cls) -> int: 117 """Get the maximum number of examples to include in the prompt. 118 119 Returns: 120 int: The maximum number of examples (default 25). 121 """ 122 return 25
Get the maximum number of examples to include in the prompt.
Returns: int: The maximum number of examples (default 25).
124 def build_base_prompt(self) -> str: 125 """Build a prompt with instruction, requirements, and multiple examples. 126 127 Returns: 128 str: The constructed prompt string with examples. 129 """ 130 base_prompt = f"# Instruction\n\n{self.task.instruction}\n\n" 131 132 if len(self.task.requirements) > 0: 133 base_prompt += "# Requirements\n\nYour response should respect the following requirements:\n" 134 for i, requirement in enumerate(self.task.requirements): 135 base_prompt += f"{i + 1}) {requirement.instruction}\n" 136 base_prompt += "\n" 137 138 valid_examples = self.collect_examples() 139 140 if len(valid_examples) == 0: 141 return base_prompt 142 143 base_prompt += "# Example Outputs\n\n" 144 for i, example in enumerate(valid_examples): 145 base_prompt += self.prompt_section_for_example(i, example) 146 147 return base_prompt
Build a prompt with instruction, requirements, and multiple examples.
Returns: str: The constructed prompt string with examples.
149 def prompt_section_for_example(self, index: int, example: TaskRun) -> str: 150 # Prefer repaired output if it exists, otherwise use the regular output 151 output = example.repaired_output or example.output 152 return f"## Example {index + 1}\n\nInput: {example.input}\nOutput: {output.output}\n\n"
154 def collect_examples(self) -> list[TaskRun]: 155 valid_examples: list[TaskRun] = [] 156 runs = self.task.runs(readonly=True) 157 158 # first pass, we look for repaired outputs. These are the best examples. 159 for run in runs: 160 if len(valid_examples) >= self.__class__.example_count(): 161 break 162 if run.repaired_output is not None: 163 valid_examples.append(run) 164 165 # second pass, we look for high quality outputs (rating based) 166 # Minimum is "high_quality" (4 star in star rating scale), then sort by rating 167 # exclude repaired outputs as they were used above 168 runs_with_rating = [ 169 run 170 for run in runs 171 if run.output.rating is not None 172 and run.output.rating.value is not None 173 and run.output.rating.is_high_quality() 174 and run.repaired_output is None 175 ] 176 runs_with_rating.sort( 177 key=lambda x: (x.output.rating and x.output.rating.value) or 0, reverse=True 178 ) 179 for run in runs_with_rating: 180 if len(valid_examples) >= self.__class__.example_count(): 181 break 182 valid_examples.append(run) 183 return valid_examples
186class FewShotPromptBuilder(MultiShotPromptBuilder): 187 """A prompt builder that includes a small number of examples in the prompt.""" 188 189 @classmethod 190 def example_count(cls) -> int: 191 """Get the maximum number of examples to include in the prompt. 192 193 Returns: 194 int: The maximum number of examples (4). 195 """ 196 return 4
A prompt builder that includes a small number of examples in the prompt.
189 @classmethod 190 def example_count(cls) -> int: 191 """Get the maximum number of examples to include in the prompt. 192 193 Returns: 194 int: The maximum number of examples (4). 195 """ 196 return 4
Get the maximum number of examples to include in the prompt.
Returns: int: The maximum number of examples (4).
199class RepairsPromptBuilder(MultiShotPromptBuilder): 200 """A prompt builder that includes multiple examples in the prompt, including repaired instructions describing what was wrong, and how it was fixed.""" 201 202 def prompt_section_for_example(self, index: int, example: TaskRun) -> str: 203 if ( 204 not example.repaired_output 205 or not example.repair_instructions 206 or not example.repaired_output.output 207 ): 208 return super().prompt_section_for_example(index, example) 209 210 prompt_section = f"## Example {index + 1}\n\nInput: {example.input}\n\n" 211 prompt_section += ( 212 f"Initial Output Which Was Insufficient: {example.output.output}\n\n" 213 ) 214 prompt_section += f"Instructions On How to Improve the Initial Output: {example.repair_instructions}\n\n" 215 prompt_section += ( 216 f"Repaired Output Which is Sufficient: {example.repaired_output.output}\n\n" 217 ) 218 return prompt_section
A prompt builder that includes multiple examples in the prompt, including repaired instructions describing what was wrong, and how it was fixed.
202 def prompt_section_for_example(self, index: int, example: TaskRun) -> str: 203 if ( 204 not example.repaired_output 205 or not example.repair_instructions 206 or not example.repaired_output.output 207 ): 208 return super().prompt_section_for_example(index, example) 209 210 prompt_section = f"## Example {index + 1}\n\nInput: {example.input}\n\n" 211 prompt_section += ( 212 f"Initial Output Which Was Insufficient: {example.output.output}\n\n" 213 ) 214 prompt_section += f"Instructions On How to Improve the Initial Output: {example.repair_instructions}\n\n" 215 prompt_section += ( 216 f"Repaired Output Which is Sufficient: {example.repaired_output.output}\n\n" 217 ) 218 return prompt_section
221def chain_of_thought_prompt(task: Task) -> str: 222 """Standard implementation to build and return the chain of thought prompt string. 223 224 Returns: 225 str: The constructed chain of thought prompt. 226 """ 227 228 cot_instruction = task.thinking_instruction 229 if not cot_instruction: 230 cot_instruction = "Think step by step, explaining your reasoning." 231 232 return cot_instruction
Standard implementation to build and return the chain of thought prompt string.
Returns: str: The constructed chain of thought prompt.
235class SimpleChainOfThoughtPromptBuilder(SimplePromptBuilder): 236 """A prompt builder that includes a chain of thought prompt on top of the simple prompt.""" 237 238 def chain_of_thought_prompt(self) -> str | None: 239 return chain_of_thought_prompt(self.task)
A prompt builder that includes a chain of thought prompt on top of the simple prompt.
242class FewShotChainOfThoughtPromptBuilder(FewShotPromptBuilder): 243 """A prompt builder that includes a chain of thought prompt on top of the few shot prompt.""" 244 245 def chain_of_thought_prompt(self) -> str | None: 246 return chain_of_thought_prompt(self.task)
A prompt builder that includes a chain of thought prompt on top of the few shot prompt.
249class MultiShotChainOfThoughtPromptBuilder(MultiShotPromptBuilder): 250 """A prompt builder that includes a chain of thought prompt on top of the multi shot prompt.""" 251 252 def chain_of_thought_prompt(self) -> str | None: 253 return chain_of_thought_prompt(self.task)
A prompt builder that includes a chain of thought prompt on top of the multi shot prompt.
256class SavedPromptBuilder(BasePromptBuilder): 257 """A prompt builder that looks up a static prompt.""" 258 259 def __init__(self, task: Task, prompt_id: str): 260 super().__init__(task) 261 prompt_model = next( 262 ( 263 prompt 264 for prompt in task.prompts(readonly=True) 265 if prompt.id == prompt_id 266 ), 267 None, 268 ) 269 if not prompt_model: 270 raise ValueError(f"Prompt ID not found: {prompt_id}") 271 self.prompt_model = prompt_model 272 273 def prompt_id(self) -> str | None: 274 return self.prompt_model.id 275 276 def build_base_prompt(self) -> str: 277 """Returns a saved prompt. 278 279 Returns: 280 str: The prompt string. 281 """ 282 return self.prompt_model.prompt 283 284 def chain_of_thought_prompt(self) -> str | None: 285 return self.prompt_model.chain_of_thought_instructions
A prompt builder that looks up a static prompt.
259 def __init__(self, task: Task, prompt_id: str): 260 super().__init__(task) 261 prompt_model = next( 262 ( 263 prompt 264 for prompt in task.prompts(readonly=True) 265 if prompt.id == prompt_id 266 ), 267 None, 268 ) 269 if not prompt_model: 270 raise ValueError(f"Prompt ID not found: {prompt_id}") 271 self.prompt_model = prompt_model
Initialize the prompt builder with a task.
Args: task (Task): The task containing instructions and requirements.
Returns the ID of the prompt, scoped to this builder.
Returns: str | None: The ID of the prompt, or None if not set.
276 def build_base_prompt(self) -> str: 277 """Returns a saved prompt. 278 279 Returns: 280 str: The prompt string. 281 """ 282 return self.prompt_model.prompt
Returns a saved prompt.
Returns: str: The prompt string.
284 def chain_of_thought_prompt(self) -> str | None: 285 return self.prompt_model.chain_of_thought_instructions
Build and return the chain of thought prompt string.
Returns: str: The constructed chain of thought prompt.
Inherited Members
288class TaskRunConfigPromptBuilder(BasePromptBuilder): 289 """A prompt builder that looks up a static prompt in a task run config.""" 290 291 def __init__(self, task: Task, run_config_prompt_id: str): 292 parts = run_config_prompt_id.split("::") 293 if len(parts) != 4: 294 raise ValueError( 295 f"Invalid task run config prompt ID: {run_config_prompt_id}. Expected format: 'task_run_config::[project_id]::[task_id]::[run_config_id]'." 296 ) 297 298 task_id = parts[2] 299 if task_id != task.id: 300 raise ValueError( 301 f"Task run config prompt ID: {run_config_prompt_id}. Task ID mismatch. Expected: {task.id}, got: {task_id}." 302 ) 303 304 run_config_id = parts[3] 305 run_config = next( 306 ( 307 run_config 308 for run_config in task.run_configs(readonly=True) 309 if run_config.id == run_config_id 310 ), 311 None, 312 ) 313 if not run_config: 314 raise ValueError( 315 f"Task run config ID not found: {run_config_id} for prompt id {run_config_prompt_id}" 316 ) 317 if run_config.prompt is None: 318 raise ValueError( 319 f"Task run config ID {run_config_id} does not have a stored prompt. Used as prompt id {run_config_prompt_id}" 320 ) 321 322 # Load the prompt from the model 323 self.prompt = run_config.prompt.prompt 324 self.cot_prompt = run_config.prompt.chain_of_thought_instructions 325 self.id = run_config_prompt_id 326 327 super().__init__(task) 328 329 def prompt_id(self) -> str | None: 330 return self.id 331 332 def build_base_prompt(self) -> str: 333 return self.prompt 334 335 def chain_of_thought_prompt(self) -> str | None: 336 return self.cot_prompt
A prompt builder that looks up a static prompt in a task run config.
291 def __init__(self, task: Task, run_config_prompt_id: str): 292 parts = run_config_prompt_id.split("::") 293 if len(parts) != 4: 294 raise ValueError( 295 f"Invalid task run config prompt ID: {run_config_prompt_id}. Expected format: 'task_run_config::[project_id]::[task_id]::[run_config_id]'." 296 ) 297 298 task_id = parts[2] 299 if task_id != task.id: 300 raise ValueError( 301 f"Task run config prompt ID: {run_config_prompt_id}. Task ID mismatch. Expected: {task.id}, got: {task_id}." 302 ) 303 304 run_config_id = parts[3] 305 run_config = next( 306 ( 307 run_config 308 for run_config in task.run_configs(readonly=True) 309 if run_config.id == run_config_id 310 ), 311 None, 312 ) 313 if not run_config: 314 raise ValueError( 315 f"Task run config ID not found: {run_config_id} for prompt id {run_config_prompt_id}" 316 ) 317 if run_config.prompt is None: 318 raise ValueError( 319 f"Task run config ID {run_config_id} does not have a stored prompt. Used as prompt id {run_config_prompt_id}" 320 ) 321 322 # Load the prompt from the model 323 self.prompt = run_config.prompt.prompt 324 self.cot_prompt = run_config.prompt.chain_of_thought_instructions 325 self.id = run_config_prompt_id 326 327 super().__init__(task)
Initialize the prompt builder with a task.
Args: task (Task): The task containing instructions and requirements.
Returns the ID of the prompt, scoped to this builder.
Returns: str | None: The ID of the prompt, or None if not set.
Build and return the complete prompt string.
Returns: str: The constructed prompt.
Build and return the chain of thought prompt string.
Returns: str: The constructed chain of thought prompt.
Inherited Members
339class FineTunePromptBuilder(BasePromptBuilder): 340 """A prompt builder that looks up a fine-tune prompt.""" 341 342 def __init__(self, task: Task, nested_fine_tune_id: str): 343 super().__init__(task) 344 345 # IDs are in project_id::task_id::fine_tune_id format 346 self.full_fine_tune_id = nested_fine_tune_id 347 parts = nested_fine_tune_id.split("::") 348 if len(parts) != 3: 349 raise ValueError( 350 f"Invalid fine-tune ID format. Expected 'project_id::task_id::fine_tune_id', got: {nested_fine_tune_id}" 351 ) 352 fine_tune_id = parts[2] 353 354 fine_tune_model = next( 355 ( 356 fine_tune 357 for fine_tune in task.finetunes(readonly=True) 358 if fine_tune.id == fine_tune_id 359 ), 360 None, 361 ) 362 if not fine_tune_model: 363 raise ValueError(f"Fine-tune ID not found: {fine_tune_id}") 364 self.fine_tune_model = fine_tune_model 365 366 def prompt_id(self) -> str | None: 367 return self.full_fine_tune_id 368 369 def build_base_prompt(self) -> str: 370 return self.fine_tune_model.system_message 371 372 def chain_of_thought_prompt(self) -> str | None: 373 return self.fine_tune_model.thinking_instructions
A prompt builder that looks up a fine-tune prompt.
342 def __init__(self, task: Task, nested_fine_tune_id: str): 343 super().__init__(task) 344 345 # IDs are in project_id::task_id::fine_tune_id format 346 self.full_fine_tune_id = nested_fine_tune_id 347 parts = nested_fine_tune_id.split("::") 348 if len(parts) != 3: 349 raise ValueError( 350 f"Invalid fine-tune ID format. Expected 'project_id::task_id::fine_tune_id', got: {nested_fine_tune_id}" 351 ) 352 fine_tune_id = parts[2] 353 354 fine_tune_model = next( 355 ( 356 fine_tune 357 for fine_tune in task.finetunes(readonly=True) 358 if fine_tune.id == fine_tune_id 359 ), 360 None, 361 ) 362 if not fine_tune_model: 363 raise ValueError(f"Fine-tune ID not found: {fine_tune_id}") 364 self.fine_tune_model = fine_tune_model
Initialize the prompt builder with a task.
Args: task (Task): The task containing instructions and requirements.
Returns the ID of the prompt, scoped to this builder.
Returns: str | None: The ID of the prompt, or None if not set.
Build and return the complete prompt string.
Returns: str: The constructed prompt.
372 def chain_of_thought_prompt(self) -> str | None: 373 return self.fine_tune_model.thinking_instructions
Build and return the chain of thought prompt string.
Returns: str: The constructed chain of thought prompt.
Inherited Members
377def prompt_builder_from_id(prompt_id: PromptId, task: Task) -> BasePromptBuilder: 378 """Convert a name used in the UI to the corresponding prompt builder class. 379 380 Args: 381 prompt_id (PromptId): The prompt ID. 382 383 Returns: 384 type[BasePromptBuilder]: The corresponding prompt builder class. 385 386 Raises: 387 ValueError: If the UI name is not recognized. 388 """ 389 390 # Saved prompts are prefixed with "id::" 391 if prompt_id.startswith("id::"): 392 prompt_id = prompt_id[4:] 393 return SavedPromptBuilder(task, prompt_id) 394 395 # Task run config prompts are prefixed with "task_run_config::" 396 # task_run_config::[project_id]::[task_id]::[run_config_id] 397 if prompt_id.startswith("task_run_config::"): 398 return TaskRunConfigPromptBuilder(task, prompt_id) 399 400 # Fine-tune prompts are prefixed with "fine_tune_prompt::" 401 if prompt_id.startswith("fine_tune_prompt::"): 402 prompt_id = prompt_id[18:] 403 return FineTunePromptBuilder(task, prompt_id) 404 405 # Check if the prompt_id matches any enum value 406 if prompt_id not in [member.value for member in PromptGenerators]: 407 raise ValueError(f"Unknown prompt generator: {prompt_id}") 408 typed_prompt_generator = PromptGenerators(prompt_id) 409 410 match typed_prompt_generator: 411 case PromptGenerators.SIMPLE: 412 return SimplePromptBuilder(task) 413 case PromptGenerators.SHORT: 414 return ShortPromptBuilder(task) 415 case PromptGenerators.FEW_SHOT: 416 return FewShotPromptBuilder(task) 417 case PromptGenerators.MULTI_SHOT: 418 return MultiShotPromptBuilder(task) 419 case PromptGenerators.REPAIRS: 420 return RepairsPromptBuilder(task) 421 case PromptGenerators.SIMPLE_CHAIN_OF_THOUGHT: 422 return SimpleChainOfThoughtPromptBuilder(task) 423 case PromptGenerators.FEW_SHOT_CHAIN_OF_THOUGHT: 424 return FewShotChainOfThoughtPromptBuilder(task) 425 case PromptGenerators.MULTI_SHOT_CHAIN_OF_THOUGHT: 426 return MultiShotChainOfThoughtPromptBuilder(task) 427 case _: 428 # Type checking will find missing cases 429 raise_exhaustive_enum_error(typed_prompt_generator)
Convert a name used in the UI to the corresponding prompt builder class.
Args: prompt_id (PromptId): The prompt ID.
Returns: type[BasePromptBuilder]: The corresponding prompt builder class.
Raises: ValueError: If the UI name is not recognized.