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)
class BasePromptBuilder:
 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.

BasePromptBuilder(task: kiln_ai.datamodel.Task)
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.

task
def prompt_id(self) -> str | None:
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.

def build_prompt(self, include_json_instructions) -> str:
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.

@abstractmethod
def build_base_prompt(self) -> str:
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.

def chain_of_thought_prompt(self) -> str | None:
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.

def build_prompt_for_ui(self) -> str:
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.

class SimplePromptBuilder(BasePromptBuilder):
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.

def build_base_prompt(self) -> str:
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.

class ShortPromptBuilder(BasePromptBuilder):
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.

def build_base_prompt(self) -> str:
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.

class MultiShotPromptBuilder(BasePromptBuilder):
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.

@classmethod
def example_count(cls) -> int:
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).

def build_base_prompt(self) -> str:
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.

def prompt_section_for_example(self, index: int, example: kiln_ai.datamodel.TaskRun) -> str:
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"
def collect_examples(self) -> list[kiln_ai.datamodel.TaskRun]:
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
class FewShotPromptBuilder(MultiShotPromptBuilder):
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.

@classmethod
def example_count(cls) -> int:
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).

class RepairsPromptBuilder(MultiShotPromptBuilder):
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.

def prompt_section_for_example(self, index: int, example: kiln_ai.datamodel.TaskRun) -> str:
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
def chain_of_thought_prompt(task: kiln_ai.datamodel.Task) -> str:
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.

class SimpleChainOfThoughtPromptBuilder(SimplePromptBuilder):
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.

def chain_of_thought_prompt(self) -> str | None:
238    def chain_of_thought_prompt(self) -> str | None:
239        return chain_of_thought_prompt(self.task)

Build and return the chain of thought prompt string.

Returns: str: The constructed chain of thought prompt.

class FewShotChainOfThoughtPromptBuilder(FewShotPromptBuilder):
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.

def chain_of_thought_prompt(self) -> str | None:
245    def chain_of_thought_prompt(self) -> str | None:
246        return chain_of_thought_prompt(self.task)

Build and return the chain of thought prompt string.

Returns: str: The constructed chain of thought prompt.

class MultiShotChainOfThoughtPromptBuilder(MultiShotPromptBuilder):
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.

def chain_of_thought_prompt(self) -> str | None:
252    def chain_of_thought_prompt(self) -> str | None:
253        return chain_of_thought_prompt(self.task)

Build and return the chain of thought prompt string.

Returns: str: The constructed chain of thought prompt.

class SavedPromptBuilder(BasePromptBuilder):
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.

SavedPromptBuilder(task: kiln_ai.datamodel.Task, prompt_id: str)
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.

prompt_model
def prompt_id(self) -> str | None:
273    def prompt_id(self) -> str | None:
274        return self.prompt_model.id

Returns the ID of the prompt, scoped to this builder.

Returns: str | None: The ID of the prompt, or None if not set.

def build_base_prompt(self) -> str:
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.

def chain_of_thought_prompt(self) -> str | None:
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.

class TaskRunConfigPromptBuilder(BasePromptBuilder):
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.

TaskRunConfigPromptBuilder(task: kiln_ai.datamodel.Task, run_config_prompt_id: str)
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.

prompt
cot_prompt
id
def prompt_id(self) -> str | None:
329    def prompt_id(self) -> str | None:
330        return self.id

Returns the ID of the prompt, scoped to this builder.

Returns: str | None: The ID of the prompt, or None if not set.

def build_base_prompt(self) -> str:
332    def build_base_prompt(self) -> str:
333        return self.prompt

Build and return the complete prompt string.

Returns: str: The constructed prompt.

def chain_of_thought_prompt(self) -> str | None:
335    def chain_of_thought_prompt(self) -> str | None:
336        return self.cot_prompt

Build and return the chain of thought prompt string.

Returns: str: The constructed chain of thought prompt.

class FineTunePromptBuilder(BasePromptBuilder):
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.

FineTunePromptBuilder(task: kiln_ai.datamodel.Task, nested_fine_tune_id: str)
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.

full_fine_tune_id
fine_tune_model
def prompt_id(self) -> str | None:
366    def prompt_id(self) -> str | None:
367        return self.full_fine_tune_id

Returns the ID of the prompt, scoped to this builder.

Returns: str | None: The ID of the prompt, or None if not set.

def build_base_prompt(self) -> str:
369    def build_base_prompt(self) -> str:
370        return self.fine_tune_model.system_message

Build and return the complete prompt string.

Returns: str: The constructed prompt.

def chain_of_thought_prompt(self) -> str | None:
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.

def prompt_builder_from_id( prompt_id: Annotated[str, AfterValidator(func=<function <lambda>>)], task: kiln_ai.datamodel.Task) -> BasePromptBuilder:
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.