kiln_ai.adapters.repair.repair_task 

 1import json
 2from typing import Type
 3
 4from pydantic import BaseModel, Field
 5
 6from kiln_ai.adapters.prompt_builders import (
 7    BasePromptBuilder,
 8    SavedPromptBuilder,
 9    prompt_builder_from_id,
10)
11from kiln_ai.datamodel import Priority, Project, Task, TaskRequirement, TaskRun
12
13
14# TODO add evaluator rating
15class RepairTaskInput(BaseModel):
16    original_prompt: str
17    original_input: str
18    original_output: str
19    evaluator_feedback: str = Field(
20        min_length=1,
21        description="Feedback from an evaluator on how to repair the task run.",
22    )
23
24
25class RepairTaskRun(Task, parent_of={}):
26    def __init__(self, original_task: Task):
27        # Keep the typechecker happy
28        tmp_project = Project(name="Repair")
29        super().__init__(
30            name="Repair",
31            parent=tmp_project,
32            description="Repair a task run, given feedback from an evaluator about how the response can be improved.",
33            instruction="You are an assistant which helps improve output from another assistant (original assistant). You'll be provided a task that the original assistant executed (prompt), \
34the input it was given, and the output it generated. An evaluator has determined that the output it generated did not satisfy the task and should be improved. The evaluator will provide \
35feedback describing what should be improved. Your job is to understand the evaluator's feedback and improve the response.",
36            requirements=[
37                TaskRequirement(
38                    name="Follow Eval Feedback",
39                    instruction="The evaluator's feedback is the most important thing to consider. If it conflicts with the original task instruction or prompt, prioritize the evaluator's feedback.",
40                    priority=Priority.p0,
41                )
42            ],
43            input_json_schema=json.dumps(RepairTaskInput.model_json_schema()),
44            output_json_schema=original_task.output_json_schema,
45        )
46
47    @classmethod
48    def _original_prompt(cls, run: TaskRun, task: Task) -> str:
49        if run.output.source is None or run.output.source.properties is None:
50            raise ValueError("No source properties found")
51
52        # Get the prompt builder id. Need the second check because we used to store this in a prompt_builder_name field, so loading legacy runs will need this.
53        prompt_id = run.output.source.properties.get(
54            "prompt_id"
55        ) or run.output.source.properties.get("prompt_builder_name", None)
56        if prompt_id is not None and isinstance(prompt_id, str):
57            prompt_builder = prompt_builder_from_id(prompt_id, task)
58            if isinstance(prompt_builder, BasePromptBuilder):
59                return prompt_builder.build_prompt(include_json_instructions=False)
60
61        raise ValueError(f"Prompt builder '{prompt_id}' is not a valid prompt builder")
62
63    @classmethod
64    def build_repair_task_input(
65        cls, original_task: Task, task_run: TaskRun, evaluator_feedback: str
66    ) -> RepairTaskInput:
67        original_prompt = cls._original_prompt(task_run, original_task)
68        return RepairTaskInput(
69            original_prompt=original_prompt,
70            original_input=task_run.input,
71            original_output=task_run.output.output,
72            evaluator_feedback=evaluator_feedback,
73        )
class RepairTaskInput(pydantic.main.BaseModel): 
16class RepairTaskInput(BaseModel):
17    original_prompt: str
18    original_input: str
19    original_output: str
20    evaluator_feedback: str = Field(
21        min_length=1,
22        description="Feedback from an evaluator on how to repair the task run.",
23    )

Usage docs: https://docs.pydantic.dev/2.10/concepts/models/

A base class for creating Pydantic models.

Attributes: __class_vars__: The names of the class variables defined on the model. __private_attributes__: Metadata about the private attributes of the model. __signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.

__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom `__init__` function.
__pydantic_decorators__: Metadata containing the decorators defined on the model.
    This replaces `Model.__validators__` and `Model.__root_validators__` from Pydantic V1.
__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to
    __args__, __origin__, __parameters__ in typing-module generics. May eventually be replaced by these.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [`RootModel`][pydantic.root_model.RootModel].
__pydantic_serializer__: The `pydantic-core` `SchemaSerializer` used to dump instances of the model.
__pydantic_validator__: The `pydantic-core` `SchemaValidator` used to validate instances of the model.

__pydantic_fields__: A dictionary of field names and their corresponding [`FieldInfo`][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [`ComputedFieldInfo`][pydantic.fields.ComputedFieldInfo] objects.

__pydantic_extra__: A dictionary containing extra values, if [`extra`][pydantic.config.ConfigDict.extra]
    is set to `'allow'`.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.
original_prompt: str
original_input: str
original_output: str
evaluator_feedback: str
model_config: ClassVar[pydantic.config.ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class RepairTaskRun(kiln_ai.datamodel.task.Task): 
26class RepairTaskRun(Task, parent_of={}):
27    def __init__(self, original_task: Task):
28        # Keep the typechecker happy
29        tmp_project = Project(name="Repair")
30        super().__init__(
31            name="Repair",
32            parent=tmp_project,
33            description="Repair a task run, given feedback from an evaluator about how the response can be improved.",
34            instruction="You are an assistant which helps improve output from another assistant (original assistant). You'll be provided a task that the original assistant executed (prompt), \
35the input it was given, and the output it generated. An evaluator has determined that the output it generated did not satisfy the task and should be improved. The evaluator will provide \
36feedback describing what should be improved. Your job is to understand the evaluator's feedback and improve the response.",
37            requirements=[
38                TaskRequirement(
39                    name="Follow Eval Feedback",
40                    instruction="The evaluator's feedback is the most important thing to consider. If it conflicts with the original task instruction or prompt, prioritize the evaluator's feedback.",
41                    priority=Priority.p0,
42                )
43            ],
44            input_json_schema=json.dumps(RepairTaskInput.model_json_schema()),
45            output_json_schema=original_task.output_json_schema,
46        )
47
48    @classmethod
49    def _original_prompt(cls, run: TaskRun, task: Task) -> str:
50        if run.output.source is None or run.output.source.properties is None:
51            raise ValueError("No source properties found")
52
53        # Get the prompt builder id. Need the second check because we used to store this in a prompt_builder_name field, so loading legacy runs will need this.
54        prompt_id = run.output.source.properties.get(
55            "prompt_id"
56        ) or run.output.source.properties.get("prompt_builder_name", None)
57        if prompt_id is not None and isinstance(prompt_id, str):
58            prompt_builder = prompt_builder_from_id(prompt_id, task)
59            if isinstance(prompt_builder, BasePromptBuilder):
60                return prompt_builder.build_prompt(include_json_instructions=False)
61
62        raise ValueError(f"Prompt builder '{prompt_id}' is not a valid prompt builder")
63
64    @classmethod
65    def build_repair_task_input(
66        cls, original_task: Task, task_run: TaskRun, evaluator_feedback: str
67    ) -> RepairTaskInput:
68        original_prompt = cls._original_prompt(task_run, original_task)
69        return RepairTaskInput(
70            original_prompt=original_prompt,
71            original_input=task_run.input,
72            original_output=task_run.output.output,
73            evaluator_feedback=evaluator_feedback,
74        )

Represents a specific task to be performed, with associated requirements and validation rules.

Contains the task definition, requirements, input/output schemas, and maintains a collection of task runs.

RepairTaskRun(original_task: kiln_ai.datamodel.Task) 
27    def __init__(self, original_task: Task):
28        # Keep the typechecker happy
29        tmp_project = Project(name="Repair")
30        super().__init__(
31            name="Repair",
32            parent=tmp_project,
33            description="Repair a task run, given feedback from an evaluator about how the response can be improved.",
34            instruction="You are an assistant which helps improve output from another assistant (original assistant). You'll be provided a task that the original assistant executed (prompt), \
35the input it was given, and the output it generated. An evaluator has determined that the output it generated did not satisfy the task and should be improved. The evaluator will provide \
36feedback describing what should be improved. Your job is to understand the evaluator's feedback and improve the response.",
37            requirements=[
38                TaskRequirement(
39                    name="Follow Eval Feedback",
40                    instruction="The evaluator's feedback is the most important thing to consider. If it conflicts with the original task instruction or prompt, prioritize the evaluator's feedback.",
41                    priority=Priority.p0,
42                )
43            ],
44            input_json_schema=json.dumps(RepairTaskInput.model_json_schema()),
45            output_json_schema=original_task.output_json_schema,
46        )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

@classmethod
def build_repair_task_input( cls, original_task: kiln_ai.datamodel.Task, task_run: kiln_ai.datamodel.TaskRun, evaluator_feedback: str) -> RepairTaskInput: 
64    @classmethod
65    def build_repair_task_input(
66        cls, original_task: Task, task_run: TaskRun, evaluator_feedback: str
67    ) -> RepairTaskInput:
68        original_prompt = cls._original_prompt(task_run, original_task)
69        return RepairTaskInput(
70            original_prompt=original_prompt,
71            original_input=task_run.input,
72            original_output=task_run.output.output,
73            evaluator_feedback=evaluator_feedback,
74        )
model_config = {'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

def model_post_init(self: pydantic.main.BaseModel, context: Any, /) -> None: 
122                    def wrapped_model_post_init(self: BaseModel, context: Any, /) -> None:
123                        """We need to both initialize private attributes and call the user-defined model_post_init
124                        method.
125                        """
126                        init_private_attributes(self, context)
127                        original_model_post_init(self, context)

We need to both initialize private attributes and call the user-defined model_post_init method.