kiln_ai.datamodel

See our docs for details about our datamodel classes and hierarchy:

Developer docs: https://kiln-ai.github.io/Kiln/kiln_core_docs/kiln_ai.html

User docs: https://docs.kiln.tech/developers/kiln-datamodel

View Source

  1"""
  2See our docs for details about our datamodel classes and hierarchy:
  3
  4Developer docs: https://kiln-ai.github.io/Kiln/kiln_core_docs/kiln_ai.html
  5
  6User docs: https://docs.kiln.tech/developers/kiln-datamodel
  7"""
  8
  9# This component uses "flat" imports so we don't have too much internal structure exposed in the API.
 10# for example you can just `from datamodel import Task, Project` instead of `from datamodel.task import Task; from datamodel.project import Project`
 11
 12from __future__ import annotations
 13
 14from kiln_ai.datamodel import (
 15    chunk,
 16    dataset_split,
 17    embedding,
 18    eval,
 19    extraction,
 20    rag,
 21    reranker,
 22    strict_mode,
 23)
 24from kiln_ai.datamodel.basemodel import generate_model_id
 25from kiln_ai.datamodel.claim_review import ClaimReview, GradedClaim
 26from kiln_ai.datamodel.code_tool import CodeTool
 27from kiln_ai.datamodel.data_guide import DataGuide
 28from kiln_ai.datamodel.datamodel_enums import (
 29    FeedbackSource,
 30    FineTuneStatusType,
 31    Priority,
 32    StructuredOutputMode,
 33    TaskOutputRatingType,
 34)
 35from kiln_ai.datamodel.dataset_split import DatasetSplit, DatasetSplitDefinition
 36from kiln_ai.datamodel.external_tool_server import ExternalToolServer
 37from kiln_ai.datamodel.feedback import Feedback
 38from kiln_ai.datamodel.finetune import Finetune
 39from kiln_ai.datamodel.project import Project
 40from kiln_ai.datamodel.prompt import BasePrompt, Prompt
 41from kiln_ai.datamodel.prompt_id import (
 42    PromptGenerators,
 43    PromptId,
 44    prompt_generator_values,
 45)
 46from kiln_ai.datamodel.prompt_optimization_job import PromptOptimizationJob
 47from kiln_ai.datamodel.skill import Skill
 48from kiln_ai.datamodel.task import Task, TaskRequirement
 49from kiln_ai.datamodel.task_output import (
 50    DataSource,
 51    DataSourceProperty,
 52    DataSourceType,
 53    RequirementRating,
 54    TaskOutput,
 55    TaskOutputRating,
 56)
 57from kiln_ai.datamodel.task_run import (
 58    EvalItemSource,
 59    MessageUsage,
 60    TaskRun,
 61    Usage,
 62)
 63
 64__all__ = [
 65    "BasePrompt",
 66    "ClaimReview",
 67    "CodeTool",
 68    "DataGuide",
 69    "DataSource",
 70    "DataSourceProperty",
 71    "DataSourceType",
 72    "DatasetSplit",
 73    "DatasetSplitDefinition",
 74    "EvalItemSource",
 75    "ExternalToolServer",
 76    "Feedback",
 77    "FeedbackSource",
 78    "FineTuneStatusType",
 79    "Finetune",
 80    "GradedClaim",
 81    "MessageUsage",
 82    "Priority",
 83    "Project",
 84    "Prompt",
 85    "PromptGenerators",
 86    "PromptId",
 87    "PromptOptimizationJob",
 88    "RequirementRating",
 89    "Skill",
 90    "StructuredOutputMode",
 91    "Task",
 92    "TaskOutput",
 93    "TaskOutputRating",
 94    "TaskOutputRatingType",
 95    "TaskRequirement",
 96    "TaskRun",
 97    "Usage",
 98    "chunk",
 99    "dataset_split",
100    "embedding",
101    "eval",
102    "extraction",
103    "generate_model_id",
104    "prompt_generator_values",
105    "rag",
106    "reranker",
107    "strict_mode",
108]

class BasePrompt(pydantic.main.BaseModel): View Source

 7class BasePrompt(BaseModel):
 8    """
 9    A prompt for a task. This is the basic data storage format which can be used throughout a project.
10
11    The "Prompt" model name is reserved for the custom prompts parented by a task.
12    """
13
14    name: FilenameString = Field(description="The name of the prompt.")
15    description: str | None = Field(
16        default=None,
17        description="A more detailed description of the prompt.",
18    )
19    generator_id: str | None = Field(
20        default=None,
21        description="The id of the generator that created this prompt.",
22    )
23    prompt: str = Field(
24        description="The prompt for the task.",
25        min_length=1,
26    )
27    chain_of_thought_instructions: str | None = Field(
28        default=None,
29        description="Instructions for the model 'thinking' about the requirement prior to answering. Used for chain of thought style prompting. COT will not be used unless this is provided.",
30    )

A prompt for a task. This is the basic data storage format which can be used throughout a project.

The "Prompt" model name is reserved for the custom prompts parented by a task.

name: Annotated[str, BeforeValidator(func=<function name_validator.<locals>.fn at 0x7f5eb7c19e40>, json_schema_input_type=PydanticUndefined), StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=120, pattern=None, ascii_only=None)]

description: str | None

generator_id: str | None

prompt: str

chain_of_thought_instructions: str | None

model_config: ClassVar[pydantic.config.ConfigDict] = {}

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

class ClaimReview(kiln_ai.datamodel.basemodel.KilnParentedModel): View Source

28class ClaimReview(KilnParentedModel):
29    """A human's grades on the claim summary of one task run.
30
31    Persisted alongside the run's rating so consumers (e.g. judge-prompt
32    refinement) can use the full review, which claims were agreed or
33    disagreed with and why, not just the final pass/fail. Every claim the
34    reviewer saw is recorded with its grade, and the overview is kept so a
35    stored review reads on its own.
36    """
37
38    judge_score: Literal["pass", "fail"] = Field(
39        description="The judge's binary verdict on this run."
40    )
41    judge_reasoning: str = Field(description="The judge's explanation for its verdict.")
42    overview: str = Field(
43        description="The neutral summary of the run the reviewer read before "
44        "grading the claims."
45    )
46    claims: list[GradedClaim] = Field(
47        description="Every graded claim, in the order the reviewer saw them.",
48    )
49    human_verdict: Literal["pass", "fail"] = Field(
50        description="The reviewer's overall call on this run. Derived from "
51        "their grade on the verdict claim when the summary carried one, "
52        "otherwise asked directly."
53    )

A human's grades on the claim summary of one task run.

Persisted alongside the run's rating so consumers (e.g. judge-prompt refinement) can use the full review, which claims were agreed or disagreed with and why, not just the final pass/fail. Every claim the reviewer saw is recorded with its grade, and the overview is kept so a stored review reads on its own.

judge_score: Literal['pass', 'fail']

judge_reasoning: str

overview: str

claims: list[GradedClaim]

human_verdict: Literal['pass', 'fail']

def relationship_name() -> str: View Source

859        def relationship_name_method() -> str:
860            return filesystem_name

The type of the None singleton.

def parent_type() -> Type[kiln_ai.datamodel.basemodel.KilnParentModel]: View Source

852        def parent_class_method() -> Type[KilnParentModel]:
853            return cls

The type of the None singleton.

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: View Source

365def init_private_attributes(self: BaseModel, context: Any, /) -> None:
366    """This function is meant to behave like a BaseModel method to initialize private attributes.
367
368    It takes context as an argument since that's what pydantic-core passes when calling it.
369
370    Args:
371        self: The BaseModel instance.
372        context: The context.
373    """
374    if getattr(self, '__pydantic_private__', None) is None:
375        pydantic_private = {}
376        for name, private_attr in self.__private_attributes__.items():
377            # Avoid needlessly creating a new dict for the validated data:
378            if private_attr.default_factory_takes_validated_data:
379                default = private_attr.get_default(
380                    call_default_factory=True, validated_data={**self.__dict__, **pydantic_private}
381                )
382            else:
383                default = private_attr.get_default(call_default_factory=True)
384            if default is not PydanticUndefined:
385                pydantic_private[name] = default
386        object_setattr(self, '__pydantic_private__', pydantic_private)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Args: self: The BaseModel instance. context: The context.

class CodeTool(kiln_ai.datamodel.basemodel.KilnParentedModel): View Source

 39class CodeTool(KilnParentedModel):
 40    """A user-authored Python function that runs as a tool inside the agent harness.
 41
 42    Functional content (code, schema, allowlist, etc.) is immutable post-create
 43    — the API enforces this; changing code means cloning into a new tool.
 44    """
 45
 46    # Editable metadata
 47    name: FilenameString = Field(description="User-facing display name.")
 48    description: str | None = Field(
 49        default=None,
 50        description="User-facing notes shown in the UI. Not shown to models.",
 51    )
 52    is_archived: bool = Field(
 53        default=False,
 54        description="Archived tools are hidden from pickers but still resolve if referenced.",
 55    )
 56
 57    # Functional content — immutable post-create (enforced at the API layer)
 58    tool_function_name: str = Field(
 59        description="The function name exposed to the model. Snake_case identifier."
 60    )
 61    tool_description: str = Field(
 62        min_length=1,
 63        description="Shown to agents as the tool description.",
 64    )
 65    parameters_schema: dict[str, Any] = Field(
 66        description="JSON Schema for the tool's parameters. Root must be type: object.",
 67    )
 68    code: str = Field(
 69        description="Python source, stored in a sibling tool.py file (in memory as a string). Validated for syntax and entry-point presence.",
 70    )
 71    timeout_seconds: int = Field(
 72        default=60,
 73        ge=1,
 74        description="Wall-clock timeout for one invocation, including nested tool calls.",
 75    )
 76    tool_allowlist: list[ToolId] = Field(
 77        default_factory=list,
 78        description="Explicit per-tool allowlist of tools this code tool may call.",
 79    )
 80
 81    @model_validator(mode="before")
 82    @classmethod
 83    def _read_code_file(cls, data: Any, info: ValidationInfo) -> Any:
 84        """When loading from disk, inject `code` from the sibling tool.py.
 85
 86        The source is stored in tool.py beside code_tool.kiln, not inline in the
 87        JSON. On load the base model puts the artifact folder in the validation
 88        context (`source_dir`); the shared helper reads the file here, before
 89        field validation, so the existing validate_code trio runs against the
 90        loaded string unchanged.
 91        """
 92        return read_code_from_sibling_file(
 93            data,
 94            info.context or {},
 95            filename=TOOL_CODE_FILENAME,
 96            kiln_filename="code_tool.kiln",
 97            model_label="CodeTool",
 98        )
 99
100    @model_serializer(mode="wrap")
101    def _serialize(
102        self, handler: SerializerFunctionWrapHandler, info: SerializationInfo
103    ) -> dict[str, Any]:
104        """On disk-save, write `code` to tool.py and omit it from the .kiln JSON.
105
106        Delegates to the shared sibling-file helper, which uses the same save
107        context attachments use (`save_attachments` + `dest_path`). Without that
108        context — normal model_dump / API responses — `code` is left in the
109        output and no file is written, so the API contract is unchanged.
110
111        Trade-off: a custom model_serializer collapses the *serialization-mode*
112        JSON schema to an untyped object (`model_json_schema(mode="serialization")`
113        loses per-field typing). This is acceptable and consistent with the
114        existing KilnAttachmentModel precedent, which uses the same pattern:
115        - Validation-mode schema is unaffected, so request bodies stay fully typed.
116        - No endpoint uses `response_model=CodeTool`; every code-tool endpoint
117          returns a dedicated response model, and the generated web schema never
118          references CodeTool's serialization schema.
119        If a typed serialization schema is ever needed off this model, add a
120        `__get_pydantic_json_schema__` override rather than removing this
121        serializer. (CodeEvalProperties keeps exactly such an override because it
122        IS a FastAPI response_model member.)
123        """
124        return write_code_to_sibling_file(
125            handler(self),
126            info.context or {},
127            filename=TOOL_CODE_FILENAME,
128            code=self.code,
129        )
130
131    @field_validator("tool_function_name")
132    @classmethod
133    def validate_function_name(cls, v: str) -> str:
134        if not _FUNCTION_NAME_RE.fullmatch(v):
135            raise ValueError(
136                f"tool_function_name must match ^[a-z][a-z0-9_]{{0,63}}$, got: '{v}'"
137            )
138        return v
139
140    @model_validator(mode="after")
141    def validate_parameters_schema(self, info: ValidationInfo) -> Self:
142        validate_schema_dict(self.parameters_schema, require_object=True)
143
144        # Parameters are passed to run() as keyword arguments keyed by these names,
145        # and a Python reserved word can never be declared as a run() parameter, so
146        # reject it at authoring. Loaded files are exempt so old projects still open.
147        # Soft keywords (match/case/type/_) are valid identifiers and stay allowed.
148        if not self.loaded_from_file(info):
149            for param_name in self.parameters_schema.get("properties", {}):
150                if keyword.iskeyword(param_name):
151                    raise ValueError(
152                        f"Parameter name '{param_name}' is a reserved Python keyword "
153                        "and cannot be used as a run() parameter. Choose a different name."
154                    )
155        return self
156
157    @model_validator(mode="after")
158    def validate_code(self) -> Self:
159        code_bytes = self.code.encode("utf-8")
160        if len(code_bytes) > 64 * 1024:
161            raise ValueError(
162                f"Code is too large ({len(code_bytes)} bytes). Maximum size is 64KB."
163            )
164
165        try:
166            compile(self.code, "<code_tool>", "exec")
167        except SyntaxError as e:
168            raise ValueError(f"Code has a syntax error: {e}") from e
169
170        tree = ast.parse(self.code)
171        has_run_fn = any(
172            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
173            and node.name == "run"
174            for node in ast.iter_child_nodes(tree)
175        )
176        if not has_run_fn:
177            raise ValueError(
178                "Code must define a module-level 'run' function (def run(...) or async def run(...))."
179            )
180
181        return self
182
183    @model_validator(mode="after")
184    def validate_allowlist(self) -> Self:
185        validate_tool_allowlist(
186            self.tool_allowlist,
187            caller="code tools",
188            self_tool_id=build_code_tool_id(self.id) if self.id is not None else None,
189        )
190        return self

A user-authored Python function that runs as a tool inside the agent harness.

Functional content (code, schema, allowlist, etc.) is immutable post-create — the API enforces this; changing code means cloning into a new tool.

description: str | None

is_archived: bool

tool_function_name: str

tool_description: str

parameters_schema: dict[str, typing.Any]

code: str

timeout_seconds: int

tool_allowlist: list[typing.Annotated[str, AfterValidator(func=<function <lambda> at 0x7f5eb9365b20>)]]

@field_validator('tool_function_name')

@classmethod

def validate_function_name(cls, v: str) -> str: View Source

131    @field_validator("tool_function_name")
132    @classmethod
133    def validate_function_name(cls, v: str) -> str:
134        if not _FUNCTION_NAME_RE.fullmatch(v):
135            raise ValueError(
136                f"tool_function_name must match ^[a-z][a-z0-9_]{{0,63}}$, got: '{v}'"
137            )
138        return v

@model_validator(mode='after')

def validate_parameters_schema(self, info: pydantic_core.core_schema.ValidationInfo) -> Self: View Source

140    @model_validator(mode="after")
141    def validate_parameters_schema(self, info: ValidationInfo) -> Self:
142        validate_schema_dict(self.parameters_schema, require_object=True)
143
144        # Parameters are passed to run() as keyword arguments keyed by these names,
145        # and a Python reserved word can never be declared as a run() parameter, so
146        # reject it at authoring. Loaded files are exempt so old projects still open.
147        # Soft keywords (match/case/type/_) are valid identifiers and stay allowed.
148        if not self.loaded_from_file(info):
149            for param_name in self.parameters_schema.get("properties", {}):
150                if keyword.iskeyword(param_name):
151                    raise ValueError(
152                        f"Parameter name '{param_name}' is a reserved Python keyword "
153                        "and cannot be used as a run() parameter. Choose a different name."
154                    )
155        return self

@model_validator(mode='after')

def validate_code(self) -> Self: View Source

157    @model_validator(mode="after")
158    def validate_code(self) -> Self:
159        code_bytes = self.code.encode("utf-8")
160        if len(code_bytes) > 64 * 1024:
161            raise ValueError(
162                f"Code is too large ({len(code_bytes)} bytes). Maximum size is 64KB."
163            )
164
165        try:
166            compile(self.code, "<code_tool>", "exec")
167        except SyntaxError as e:
168            raise ValueError(f"Code has a syntax error: {e}") from e
169
170        tree = ast.parse(self.code)
171        has_run_fn = any(
172            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
173            and node.name == "run"
174            for node in ast.iter_child_nodes(tree)
175        )
176        if not has_run_fn:
177            raise ValueError(
178                "Code must define a module-level 'run' function (def run(...) or async def run(...))."
179            )
180
181        return self

@model_validator(mode='after')

def validate_allowlist(self) -> Self: View Source

183    @model_validator(mode="after")
184    def validate_allowlist(self) -> Self:
185        validate_tool_allowlist(
186            self.tool_allowlist,
187            caller="code tools",
188            self_tool_id=build_code_tool_id(self.id) if self.id is not None else None,
189        )
190        return self

def relationship_name() -> str: View Source

859        def relationship_name_method() -> str:
860            return filesystem_name

The type of the None singleton.

def parent_type() -> Type[kiln_ai.datamodel.basemodel.KilnParentModel]: View Source

852        def parent_class_method() -> Type[KilnParentModel]:
853            return cls

The type of the None singleton.

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: View Source

365def init_private_attributes(self: BaseModel, context: Any, /) -> None:
366    """This function is meant to behave like a BaseModel method to initialize private attributes.
367
368    It takes context as an argument since that's what pydantic-core passes when calling it.
369
370    Args:
371        self: The BaseModel instance.
372        context: The context.
373    """
374    if getattr(self, '__pydantic_private__', None) is None:
375        pydantic_private = {}
376        for name, private_attr in self.__private_attributes__.items():
377            # Avoid needlessly creating a new dict for the validated data:
378            if private_attr.default_factory_takes_validated_data:
379                default = private_attr.get_default(
380                    call_default_factory=True, validated_data={**self.__dict__, **pydantic_private}
381                )
382            else:
383                default = private_attr.get_default(call_default_factory=True)
384            if default is not PydanticUndefined:
385                pydantic_private[name] = default
386        object_setattr(self, '__pydantic_private__', pydantic_private)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Args: self: The BaseModel instance. context: The context.

class DataGuide(kiln_ai.datamodel.basemodel.KilnParentedModel): View Source

11class DataGuide(KilnParentedModel):
12    """Persistent input data guide for synthetic data generation, stored as a
13    child of a Task.
14
15    The guide describes what realistic *inputs* to this task look like — input
16    shape, format, distribution, and the kinds of values inputs contain. It is
17    consumed at the topic and input generation stages of synthetic data
18    generation, never at the output stage. Output behavior is the job of the
19    task's system prompt, not this guide.
20
21    The guide is a single markdown body. Two canonical shapes are supported
22    depending on origin:
23
24    - **Manual flow** (user-authored examples): leads with `# Reference Inputs`
25      (user-owned `## Example N` blocks), followed by `# Semantics`, `# Style`,
26      `# Presentation Defaults`.
27    - **Kiln Pro / Copilot flow** (analyze pipeline): only `# Semantics`,
28      `# Style`, `# Presentation Defaults` — the analyze prompt derives rules
29      from input documents rather than quoting them.
30
31    The metaprompter treats the whole body as one editable artifact and returns
32    a refined version on each refine pass; refine auto-detects which shape it
33    is working on by checking for a `# Reference Inputs` heading.
34    """
35
36    guide: str = Field(
37        default="",
38        description="Markdown body of the input data guide. Manual-flow guides start with `# Reference Inputs`; Kiln Pro / Copilot-flow guides have only `# Semantics`, `# Style`, `# Presentation Defaults`.",
39    )
40
41    source: DataGuideSource = Field(
42        default="manual",
43        description="Which flow created this guide. Refine + verify pipelines branch on this to choose the right metaprompter (manual = user-curated examples + edits all sections; kiln_pro = LLM-derived, refine is feedback-only surgical edits).",
44    )

Persistent input data guide for synthetic data generation, stored as a child of a Task.

The guide describes what realistic inputs to this task look like — input shape, format, distribution, and the kinds of values inputs contain. It is consumed at the topic and input generation stages of synthetic data generation, never at the output stage. Output behavior is the job of the task's system prompt, not this guide.

The guide is a single markdown body. Two canonical shapes are supported depending on origin:

Manual flow (user-authored examples): leads with # Reference Inputs (user-owned ## Example N blocks), followed by # Semantics, # Style, # Presentation Defaults.
Kiln Pro / Copilot flow (analyze pipeline): only # Semantics, # Style, # Presentation Defaults — the analyze prompt derives rules from input documents rather than quoting them.

The metaprompter treats the whole body as one editable artifact and returns a refined version on each refine pass; refine auto-detects which shape it is working on by checking for a # Reference Inputs heading.

guide: str

source: Literal['manual', 'kiln_pro']

def relationship_name() -> str: View Source

859        def relationship_name_method() -> str:
860            return filesystem_name

The type of the None singleton.

def parent_type() -> Type[kiln_ai.datamodel.basemodel.KilnParentModel]: View Source

852        def parent_class_method() -> Type[KilnParentModel]:
853            return cls

The type of the None singleton.

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: View Source

365def init_private_attributes(self: BaseModel, context: Any, /) -> None:
366    """This function is meant to behave like a BaseModel method to initialize private attributes.
367
368    It takes context as an argument since that's what pydantic-core passes when calling it.
369
370    Args:
371        self: The BaseModel instance.
372        context: The context.
373    """
374    if getattr(self, '__pydantic_private__', None) is None:
375        pydantic_private = {}
376        for name, private_attr in self.__private_attributes__.items():
377            # Avoid needlessly creating a new dict for the validated data:
378            if private_attr.default_factory_takes_validated_data:
379                default = private_attr.get_default(
380                    call_default_factory=True, validated_data={**self.__dict__, **pydantic_private}
381                )
382            else:
383                default = private_attr.get_default(call_default_factory=True)
384            if default is not PydanticUndefined:
385                pydantic_private[name] = default
386        object_setattr(self, '__pydantic_private__', pydantic_private)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Args: self: The BaseModel instance. context: The context.

class DataSource(pydantic.main.BaseModel): View Source

204class DataSource(BaseModel):
205    """
206    Represents the origin of data, either human, synthetic, file import, or tool call, with associated properties.
207
208    Properties vary based on the source type - for synthetic/tool_call sources this includes
209    model information, for human sources this includes creator information, for file imports
210    this includes file information.
211    """
212
213    type: DataSourceType = Field(description="The type of data source.")
214    properties: Dict[str, str | int | float] = Field(
215        default={},
216        description="Properties describing the data source. For synthetic things like model. For human: the human's name. For file_import: file information.",
217    )
218    run_config_id: Optional[str] = Field(
219        default=None,
220        description="The ID of the saved TaskRunConfig used to produce this data, if any. Only present when the run was initiated from a saved TaskRunConfig (e.g. via the saved-config dropdown or a tool call); runs configured ad-hoc from the run page leave this unset. Not validated against the file system: a TaskRunConfig may be deleted without invalidating historical runs that reference it.",
221    )
222    run_config: Optional[RunConfigProperties] = Field(
223        default=None,
224        description="The run config used to generate the data, if generated by a running a model in Kiln (only true for type=synthetic).",
225    )
226
227    @model_validator(mode="after")
228    def normalize_empty_run_config_id(self) -> Self:
229        # Some callers (e.g. tool wrappers reading from external properties)
230        # default missing IDs to "". Coerce to None so this field is either a
231        # real ID or unset — never an ambiguous empty string on disk.
232        if self.run_config_id == "":
233            self.run_config_id = None
234        return self
235
236    _data_source_properties = [
237        DataSourceProperty(
238            name="created_by",
239            type=str,
240            required_for=[DataSourceType.human],
241            not_allowed_for=[
242                DataSourceType.synthetic,
243                DataSourceType.file_import,
244                DataSourceType.tool_call,
245            ],
246        ),
247        DataSourceProperty(
248            name="model_name",
249            type=str,
250            required_for=[DataSourceType.synthetic],
251            not_allowed_for=[
252                DataSourceType.human,
253                DataSourceType.file_import,
254                DataSourceType.tool_call,
255            ],
256        ),
257        DataSourceProperty(
258            name="model_provider",
259            type=str,
260            required_for=[DataSourceType.synthetic],
261            not_allowed_for=[
262                DataSourceType.human,
263                DataSourceType.file_import,
264                DataSourceType.tool_call,
265            ],
266        ),
267        DataSourceProperty(
268            name="adapter_name",
269            type=str,
270            required_for=[DataSourceType.synthetic],
271            not_allowed_for=[
272                DataSourceType.human,
273                DataSourceType.file_import,
274                DataSourceType.tool_call,
275            ],
276        ),
277        DataSourceProperty(
278            # Legacy field -- allow loading from old runs, but we shouldn't be setting it.
279            name="prompt_builder_name",
280            type=str,
281            not_allowed_for=[
282                DataSourceType.human,
283                DataSourceType.file_import,
284                DataSourceType.tool_call,
285            ],
286        ),
287        DataSourceProperty(
288            # The PromptId of the prompt. Can be a saved prompt, fine-tune, generator name, etc. See PromptId type for more details.
289            name="prompt_id",
290            type=str,
291            not_allowed_for=[
292                DataSourceType.human,
293                DataSourceType.file_import,
294                DataSourceType.tool_call,
295            ],
296        ),
297        DataSourceProperty(
298            name="file_name",
299            type=str,
300            required_for=[DataSourceType.file_import],
301            not_allowed_for=[
302                DataSourceType.human,
303                DataSourceType.synthetic,
304                DataSourceType.tool_call,
305            ],
306        ),
307    ]
308
309    @model_validator(mode="after")
310    def validate_type(self) -> "DataSource":
311        if self.type not in DataSourceType:
312            raise ValueError(f"Invalid data source type: {self.type}")
313        return self
314
315    @model_validator(mode="after")
316    def validate_properties(self) -> "DataSource":
317        for prop in self._data_source_properties:
318            # Check the property type is correct
319            if prop.name in self.properties:
320                if not isinstance(self.properties[prop.name], prop.type):
321                    raise ValueError(
322                        f"'{prop.name}' must be of type {prop.type.__name__} for {self.type} data source"
323                    )
324            # Check the property is required for the data source type
325            if self.type in prop.required_for:
326                if prop.name not in self.properties:
327                    raise ValueError(
328                        f"'{prop.name}' is required for {self.type} data source"
329                    )
330            # Check the property is not allowed for the data source type
331            elif self.type in prop.not_allowed_for and prop.name in self.properties:
332                raise ValueError(
333                    f"'{prop.name}' is not allowed for {self.type} data source"
334                )
335        return self
336
337    @model_validator(mode="after")
338    def validate_no_empty_properties(self) -> Self:
339        for prop, value in self.properties.items():
340            if isinstance(value, str) and value == "":
341                raise ValueError(
342                    f"Property '{prop}' must be a non-empty string for {self.type} data source"
343                )
344        return self

Represents the origin of data, either human, synthetic, file import, or tool call, with associated properties.

Properties vary based on the source type - for synthetic/tool_call sources this includes model information, for human sources this includes creator information, for file imports this includes file information.

type: DataSourceType

properties: Dict[str, str | int | float]

run_config_id: Optional[str]

run_config: Optional[Annotated[Union[Annotated[kiln_ai.datamodel.run_config.KilnAgentRunConfigProperties, Tag(tag='kiln_agent')], Annotated[kiln_ai.datamodel.run_config.McpRunConfigProperties, Tag(tag='mcp')]], Discriminator(discriminator=<function _get_run_config_type at 0x7f5eb9302c00>, custom_error_type=None, custom_error_message=None, custom_error_context=None)]]

@model_validator(mode='after')

def normalize_empty_run_config_id(self) -> Self: View Source

227    @model_validator(mode="after")
228    def normalize_empty_run_config_id(self) -> Self:
229        # Some callers (e.g. tool wrappers reading from external properties)
230        # default missing IDs to "". Coerce to None so this field is either a
231        # real ID or unset — never an ambiguous empty string on disk.
232        if self.run_config_id == "":
233            self.run_config_id = None
234        return self

@model_validator(mode='after')

def validate_type(self) -> DataSource: View Source

309    @model_validator(mode="after")
310    def validate_type(self) -> "DataSource":
311        if self.type not in DataSourceType:
312            raise ValueError(f"Invalid data source type: {self.type}")
313        return self

@model_validator(mode='after')

def validate_properties(self) -> DataSource: View Source

315    @model_validator(mode="after")
316    def validate_properties(self) -> "DataSource":
317        for prop in self._data_source_properties:
318            # Check the property type is correct
319            if prop.name in self.properties:
320                if not isinstance(self.properties[prop.name], prop.type):
321                    raise ValueError(
322                        f"'{prop.name}' must be of type {prop.type.__name__} for {self.type} data source"
323                    )
324            # Check the property is required for the data source type
325            if self.type in prop.required_for:
326                if prop.name not in self.properties:
327                    raise ValueError(
328                        f"'{prop.name}' is required for {self.type} data source"
329                    )
330            # Check the property is not allowed for the data source type
331            elif self.type in prop.not_allowed_for and prop.name in self.properties:
332                raise ValueError(
333                    f"'{prop.name}' is not allowed for {self.type} data source"
334                )
335        return self

@model_validator(mode='after')

def validate_no_empty_properties(self) -> Self: View Source

337    @model_validator(mode="after")
338    def validate_no_empty_properties(self) -> Self:
339        for prop, value in self.properties.items():
340            if isinstance(value, str) and value == "":
341                raise ValueError(
342                    f"Property '{prop}' must be a non-empty string for {self.type} data source"
343                )
344        return self

model_config: ClassVar[pydantic.config.ConfigDict] = {}

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: View Source

365def init_private_attributes(self: BaseModel, context: Any, /) -> None:
366    """This function is meant to behave like a BaseModel method to initialize private attributes.
367
368    It takes context as an argument since that's what pydantic-core passes when calling it.
369
370    Args:
371        self: The BaseModel instance.
372        context: The context.
373    """
374    if getattr(self, '__pydantic_private__', None) is None:
375        pydantic_private = {}
376        for name, private_attr in self.__private_attributes__.items():
377            # Avoid needlessly creating a new dict for the validated data:
378            if private_attr.default_factory_takes_validated_data:
379                default = private_attr.get_default(
380                    call_default_factory=True, validated_data={**self.__dict__, **pydantic_private}
381                )
382            else:
383                default = private_attr.get_default(call_default_factory=True)
384            if default is not PydanticUndefined:
385                pydantic_private[name] = default
386        object_setattr(self, '__pydantic_private__', pydantic_private)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Args: self: The BaseModel instance. context: The context.

class DataSourceProperty(pydantic.main.BaseModel): View Source

190class DataSourceProperty(BaseModel):
191    """
192    Defines a property that can be associated with a data source.
193
194    Includes validation rules for when properties are required or not allowed
195    based on the data source type.
196    """
197
198    name: str
199    type: Type[Union[str, int, float]]
200    required_for: List[DataSourceType] = []
201    not_allowed_for: List[DataSourceType] = []

Defines a property that can be associated with a data source.

Includes validation rules for when properties are required or not allowed based on the data source type.

name: str

type: Type[Union[str, int, float]]

required_for: List[DataSourceType]

not_allowed_for: List[DataSourceType]

model_config: ClassVar[pydantic.config.ConfigDict] = {}

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

class DataSourceType(builtins.str, enum.Enum): View Source

176class DataSourceType(str, Enum):
177    """
178    The source type of a piece of data.
179
180    Human: a human created the data
181    Synthetic: a model created the data
182    """
183
184    human = "human"
185    synthetic = "synthetic"
186    file_import = "file_import"
187    tool_call = "tool_call"

The source type of a piece of data.

Human: a human created the data Synthetic: a model created the data

human = <DataSourceType.human: 'human'>

synthetic = <DataSourceType.synthetic: 'synthetic'>

file_import = <DataSourceType.file_import: 'file_import'>

tool_call = <DataSourceType.tool_call: 'tool_call'>

class DatasetSplitDefinition(pydantic.main.BaseModel): View Source

39class DatasetSplitDefinition(BaseModel):
40    """
41    A definition of a split in a dataset.
42
43    Example: name="train", description="The training set", percentage=0.8 (80% of the dataset)
44    """
45
46    name: FilenameString = Field(
47        description="The name of the dataset split definition."
48    )
49    description: str | None = Field(
50        default=None,
51        description="A description of the dataset for you and your team. Not used in training.",
52    )
53    percentage: float = Field(
54        ge=0.0,
55        le=1.0,
56        description="The percentage of the dataset that this split represents (between 0 and 1).",
57    )

A definition of a split in a dataset.

Example: name="train", description="The training set", percentage=0.8 (80% of the dataset)

description: str | None

percentage: float

model_config: ClassVar[pydantic.config.ConfigDict] = {}

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

class EvalItemSource(pydantic.main.BaseModel): View Source

31class EvalItemSource(BaseModel):
32    """The eval dataset item a TaskRun was generated for.
33
34    Not the run config — that lives on the same TaskRun at
35    `output.source.run_config_id`.
36    """
37
38    source_type: Literal["eval_input", "task_run"] = Field(
39        description="Which store the dataset item came from: an EvalInput (V2) or a TaskRun (V1-backed split)."
40    )
41    # `str`, not the usual `ID_TYPE` (`Optional[str]`): an id-less source is a trace-index
42    # key that collides with every other id-less source. Absence is already expressed by
43    # `TaskRun.eval_source` itself being None, so the inner id has no legitimate None
44    # state. `ItemKey`'s own nullability is inherited from `KilnBaseModel.id` rather than
45    # chosen, and a narrower id still satisfies it.
46    source_id: str = Field(
47        min_length=1,
48        description="The id of the dataset item this run was generated for. Interpreted within the store named by source_type — ids are only unique within a store.",
49    )

The eval dataset item a TaskRun was generated for.

Not the run config — that lives on the same TaskRun at output.source.run_config_id.

source_type: Literal['eval_input', 'task_run']

source_id: str

model_config: ClassVar[pydantic.config.ConfigDict] = {}

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

class Feedback(kiln_ai.datamodel.basemodel.KilnParentedModel): View Source

 8class Feedback(KilnParentedModel):
 9    """Feedback on a task run.
10
11    Supports multi-source feedback: different users, automated systems, and
12    different locations in the UI can each contribute independent feedback
13    entries on the same task run.
14    """
15
16    feedback: str = Field(
17        min_length=1,
18        description="Free-form text feedback on the task run.",
19    )
20    source: FeedbackSource = Field(
21        description="Where this feedback originated, e.g. 'run-page' or 'spec-feedback'.",
22    )

Feedback on a task run.

Supports multi-source feedback: different users, automated systems, and different locations in the UI can each contribute independent feedback entries on the same task run.

feedback: str

source: FeedbackSource

def relationship_name() -> str: View Source

859        def relationship_name_method() -> str:
860            return filesystem_name

The type of the None singleton.

def parent_type() -> Type[kiln_ai.datamodel.basemodel.KilnParentModel]: View Source

852        def parent_class_method() -> Type[KilnParentModel]:
853            return cls

The type of the None singleton.

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: View Source

365def init_private_attributes(self: BaseModel, context: Any, /) -> None:
366    """This function is meant to behave like a BaseModel method to initialize private attributes.
367
368    It takes context as an argument since that's what pydantic-core passes when calling it.
369
370    Args:
371        self: The BaseModel instance.
372        context: The context.
373    """
374    if getattr(self, '__pydantic_private__', None) is None:
375        pydantic_private = {}
376        for name, private_attr in self.__private_attributes__.items():
377            # Avoid needlessly creating a new dict for the validated data:
378            if private_attr.default_factory_takes_validated_data:
379                default = private_attr.get_default(
380                    call_default_factory=True, validated_data={**self.__dict__, **pydantic_private}
381                )
382            else:
383                default = private_attr.get_default(call_default_factory=True)
384            if default is not PydanticUndefined:
385                pydantic_private[name] = default
386        object_setattr(self, '__pydantic_private__', pydantic_private)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Args: self: The BaseModel instance. context: The context.

class FeedbackSource(builtins.str, enum.Enum): View Source

101class FeedbackSource(str, Enum):
102    """Where a piece of feedback originated.
103
104    This is an append-only enum: new sources can be added freely, but existing
105    values must never be removed or renamed so that older persisted data
106    continues to load.
107    """
108
109    run_page = "run-page"
110    spec_feedback = "spec-feedback"

Where a piece of feedback originated.

This is an append-only enum: new sources can be added freely, but existing values must never be removed or renamed so that older persisted data continues to load.

run_page = <FeedbackSource.run_page: 'run-page'>

spec_feedback = <FeedbackSource.spec_feedback: 'spec-feedback'>

class FineTuneStatusType(builtins.str, enum.Enum): View Source

62class FineTuneStatusType(str, Enum):
63    """
64    The status type of a fine-tune job.
65    """
66
67    unknown = "unknown"
68    pending = "pending"
69    running = "running"
70    completed = "completed"
71    failed = "failed"

The status type of a fine-tune job.

unknown = <FineTuneStatusType.unknown: 'unknown'>

pending = <FineTuneStatusType.pending: 'pending'>

running = <FineTuneStatusType.running: 'running'>

completed = <FineTuneStatusType.completed: 'completed'>

failed = <FineTuneStatusType.failed: 'failed'>

class Finetune(kiln_ai.datamodel.basemodel.KilnParentedModel): View Source

 24class Finetune(KilnParentedModel):
 25    """
 26    The Kiln fine-tune datamodel.
 27
 28    Initially holds a reference to a training job, with needed identifiers to update the status. When complete, contains the new model ID.
 29    """
 30
 31    name: FilenameString = Field(description="The name of the fine-tune.")
 32    description: str | None = Field(
 33        default=None,
 34        description="A description of the fine-tune for you and your team. Not used in training.",
 35    )
 36    structured_output_mode: StructuredOutputMode | None = Field(
 37        default=None,
 38        description="Legacy field -- replaced by run_config.structured_output_mode. The mode to use to train the model for structured output, if it was trained with structured output. We should call the tuned model with this mode if set.",
 39    )
 40    provider: str = Field(
 41        description="The provider to use for the fine-tune (e.g. 'openai')."
 42    )
 43    base_model_id: str = Field(
 44        description="The id of the base model to use for the fine-tune. This string relates to the provider's IDs for their own models, not Kiln IDs."
 45    )
 46    provider_id: str | None = Field(
 47        default=None,
 48        description="The ID of the fine-tune job on the provider's side. May not be the same as the fine_tune_model_id.",
 49    )
 50    fine_tune_model_id: str | None = Field(
 51        default=None,
 52        description="The ID of the fine-tuned model on the provider's side. May not be the same as the provider_id.",
 53    )
 54    dataset_split_id: str = Field(
 55        description="The ID of the dataset split to use for this fine-tune.",
 56    )
 57    train_split_name: str = Field(
 58        default="train",
 59        description="The name of the training split to use for this fine-tune.",
 60    )
 61    validation_split_name: str | None = Field(
 62        default=None,
 63        description="The name of the validation split to use for this fine-tune. Optional.",
 64    )
 65    parameters: dict[str, str | int | float | bool] = Field(
 66        default={},
 67        description="The parameters to use for this fine-tune. These are provider-specific.",
 68    )
 69    # These two fields are saved exactly used for training. Even if they map exactly to a custom prompt or generator, those can change, so we want to keep a record of the training prompt.
 70    system_message: str = Field(
 71        description="The system message to use for this fine-tune.",
 72    )
 73    thinking_instructions: str | None = Field(
 74        default=None,
 75        description="The thinking instructions to use for this fine-tune. Only used when data_strategy is final_and_intermediate.",
 76    )
 77    latest_status: FineTuneStatusType = Field(
 78        default=FineTuneStatusType.unknown,
 79        description="The latest known status of this fine-tune. Not updated in real time.",
 80    )
 81    properties: Dict[str, str | int | float] = Field(
 82        default={},
 83        description="Properties of the fine-tune. Different providers may use different properties.",
 84    )
 85    data_strategy: ChatStrategy = Field(
 86        default=ChatStrategy.single_turn,
 87        description="The strategy to use for training the model. 'final_only' will only train on the final response. 'final_and_intermediate' will train on the final response and intermediate outputs (chain of thought or reasoning).",
 88    )
 89    run_config: KilnAgentRunConfigProperties | None = Field(
 90        default=None,
 91        description="The run configuration for this fine-tune.",
 92    )
 93
 94    # Workaround to return typed parent without importing Task
 95    def parent_task(self) -> Union["Task", None]:
 96        if self.parent is None or self.parent.__class__.__name__ != "Task":
 97            return None
 98        return self.parent  # type: ignore
 99
100    def nested_id(self) -> str:
101        """
102        Build the nested ID for this finetune in the format: project_id::task_id::finetune_id
103        """
104        task = self.parent_task()
105        if task is None:
106            raise ValueError("Finetune must have a parent task")
107        project = task.parent_project()
108        if project is None:
109            raise ValueError("Finetune must have a parent project")
110        return f"{project.id}::{task.id}::{self.id}"
111
112    @model_validator(mode="after")
113    def validate_thinking_instructions(self) -> Self:
114        if (
115            self.thinking_instructions is not None
116            and self.data_strategy not in DATA_STRATEGIES_REQUIRED_THINKING_INSTRUCTIONS
117        ):
118            raise ValueError(
119                f"Thinking instructions can only be used when data_strategy is one of the following: {DATA_STRATEGIES_REQUIRED_THINKING_INSTRUCTIONS}"
120            )
121        if (
122            self.thinking_instructions is None
123            and self.data_strategy in DATA_STRATEGIES_REQUIRED_THINKING_INSTRUCTIONS
124        ):
125            raise ValueError(
126                f"Thinking instructions are required when data_strategy is one of the following: {DATA_STRATEGIES_REQUIRED_THINKING_INSTRUCTIONS}"
127            )
128        return self

The Kiln fine-tune datamodel.

Initially holds a reference to a training job, with needed identifiers to update the status. When complete, contains the new model ID.

description: str | None

structured_output_mode: StructuredOutputMode | None

provider: str

base_model_id: str

provider_id: str | None

fine_tune_model_id: str | None

dataset_split_id: str

train_split_name: str

validation_split_name: str | None

parameters: dict[str, str | int | float | bool]

system_message: str

thinking_instructions: str | None

latest_status: FineTuneStatusType

properties: Dict[str, str | int | float]

data_strategy: kiln_ai.datamodel.datamodel_enums.ChatStrategy

run_config: kiln_ai.datamodel.run_config.KilnAgentRunConfigProperties | None

def parent_task(self) -> Optional[Task]: View Source

95    def parent_task(self) -> Union["Task", None]:
96        if self.parent is None or self.parent.__class__.__name__ != "Task":
97            return None
98        return self.parent  # type: ignore

def nested_id(self) -> str: View Source

100    def nested_id(self) -> str:
101        """
102        Build the nested ID for this finetune in the format: project_id::task_id::finetune_id
103        """
104        task = self.parent_task()
105        if task is None:
106            raise ValueError("Finetune must have a parent task")
107        project = task.parent_project()
108        if project is None:
109            raise ValueError("Finetune must have a parent project")
110        return f"{project.id}::{task.id}::{self.id}"

Build the nested ID for this finetune in the format: project_id::task_id::finetune_id

@model_validator(mode='after')

def validate_thinking_instructions(self) -> Self: View Source

112    @model_validator(mode="after")
113    def validate_thinking_instructions(self) -> Self:
114        if (
115            self.thinking_instructions is not None
116            and self.data_strategy not in DATA_STRATEGIES_REQUIRED_THINKING_INSTRUCTIONS
117        ):
118            raise ValueError(
119                f"Thinking instructions can only be used when data_strategy is one of the following: {DATA_STRATEGIES_REQUIRED_THINKING_INSTRUCTIONS}"
120            )
121        if (
122            self.thinking_instructions is None
123            and self.data_strategy in DATA_STRATEGIES_REQUIRED_THINKING_INSTRUCTIONS
124        ):
125            raise ValueError(
126                f"Thinking instructions are required when data_strategy is one of the following: {DATA_STRATEGIES_REQUIRED_THINKING_INSTRUCTIONS}"
127            )
128        return self

def relationship_name() -> str: View Source

859        def relationship_name_method() -> str:
860            return filesystem_name

The type of the None singleton.

def parent_type() -> Type[kiln_ai.datamodel.basemodel.KilnParentModel]: View Source

852        def parent_class_method() -> Type[KilnParentModel]:
853            return cls

The type of the None singleton.

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: View Source

365def init_private_attributes(self: BaseModel, context: Any, /) -> None:
366    """This function is meant to behave like a BaseModel method to initialize private attributes.
367
368    It takes context as an argument since that's what pydantic-core passes when calling it.
369
370    Args:
371        self: The BaseModel instance.
372        context: The context.
373    """
374    if getattr(self, '__pydantic_private__', None) is None:
375        pydantic_private = {}
376        for name, private_attr in self.__private_attributes__.items():
377            # Avoid needlessly creating a new dict for the validated data:
378            if private_attr.default_factory_takes_validated_data:
379                default = private_attr.get_default(
380                    call_default_factory=True, validated_data={**self.__dict__, **pydantic_private}
381                )
382            else:
383                default = private_attr.get_default(call_default_factory=True)
384            if default is not PydanticUndefined:
385                pydantic_private[name] = default
386        object_setattr(self, '__pydantic_private__', pydantic_private)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Args: self: The BaseModel instance. context: The context.

class GradedClaim(pydantic.main.BaseModel): View Source

 9class GradedClaim(BaseModel):
10    """One claim with a human grade on it.
11
12    A claim is one decision the judge made, written so the reviewer can vote
13    on it from the card alone. Grades have one direction on every claim:
14    agree means the judge got that decision right, disagree means it got it
15    wrong. The claim text carries its own evidence and citation markers.
16    """
17
18    text: str = Field(description="The claim as shown to the reviewer.")
19    human_grade: Literal["agree", "disagree"] = Field(
20        description="The human's grade on this claim."
21    )
22    human_feedback: str | None = Field(
23        default=None,
24        description="Optional plaintext reason for the grade.",
25    )

One claim with a human grade on it.

A claim is one decision the judge made, written so the reviewer can vote on it from the card alone. Grades have one direction on every claim: agree means the judge got that decision right, disagree means it got it wrong. The claim text carries its own evidence and citation markers.

text: str

human_grade: Literal['agree', 'disagree']

human_feedback: str | None

model_config: ClassVar[pydantic.config.ConfigDict] = {}

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

class MessageUsage(pydantic.main.BaseModel): View Source

 57class MessageUsage(BaseModel):
 58    """Token usage and cost for a single LLM call or a multi-message sum.
 59
 60    Carries only the fields that are meaningfully aggregatable across
 61    messages: token counts and cost. Per-call latency lives on the
 62    individual message's ``latency_ms`` field; aggregating it across the
 63    full trace would mix latencies from different points in time, so
 64    ``MessageUsage`` does NOT carry ``total_llm_latency_ms``.
 65
 66    The :class:`Usage` subclass adds ``total_llm_latency_ms`` for the
 67    in-flight per-run accumulator that tracks how long this run spent
 68    waiting on LLM calls.
 69    """
 70
 71    input_tokens: int | None = Field(
 72        default=None,
 73        description="The number of input tokens used.",
 74        ge=0,
 75    )
 76    output_tokens: int | None = Field(
 77        default=None,
 78        description="The number of output tokens used.",
 79        ge=0,
 80    )
 81    total_tokens: int | None = Field(
 82        default=None,
 83        description="The total number of tokens used.",
 84        ge=0,
 85    )
 86    cost: float | None = Field(
 87        default=None,
 88        description="The cost in US dollars, saved at runtime (prices can change over time).",
 89        ge=0,
 90    )
 91    cached_tokens: int | None = Field(
 92        default=None,
 93        description="Number of tokens served from prompt cache. None if not reported.",
 94        ge=0,
 95    )
 96
 97    def __add__(self, other: "MessageUsage") -> "MessageUsage":
 98        """Add two MessageUsage objects together, handling None values gracefully.
 99
100        None + None = None
101        None + value = value
102        value + None = value
103        value1 + value2 = value1 + value2
104        """
105        if not isinstance(other, MessageUsage):
106            raise TypeError(f"Cannot add MessageUsage with {type(other).__name__}")
107
108        return MessageUsage(
109            input_tokens=_add_optional_int(self.input_tokens, other.input_tokens),
110            output_tokens=_add_optional_int(self.output_tokens, other.output_tokens),
111            total_tokens=_add_optional_int(self.total_tokens, other.total_tokens),
112            cost=_add_optional_float(self.cost, other.cost),
113            cached_tokens=_add_optional_int(self.cached_tokens, other.cached_tokens),
114        )
115
116    @staticmethod
117    def from_trace(
118        trace: "list[ChatCompletionMessageParam] | None",
119    ) -> "MessageUsage":
120        """Sum per-message usage across all assistant messages in a trace.
121
122        Returns MessageUsage() (all fields None) when trace is None/empty
123        or no assistant message has a `usage` field. Skips non-assistant
124        messages and messages where `usage` is missing or None. Always
125        returns a MessageUsage instance — never None.
126
127        Accepts per-message `usage` values that are either MessageUsage
128        instances (including the Usage subclass) or plain dicts (e.g.
129        from JSON round-trips); dicts are validated to MessageUsage
130        before summing.
131        """
132        total: MessageUsage = MessageUsage()
133        if not trace:
134            return total
135
136        for message in trace:
137            if not isinstance(message, dict):
138                continue
139            if message.get("role") != "assistant":
140                continue
141            raw_usage = message.get("usage")
142            if raw_usage is None:
143                continue
144            if isinstance(raw_usage, MessageUsage):
145                total = total + raw_usage
146            elif isinstance(raw_usage, dict):
147                total = total + MessageUsage.model_validate(raw_usage)
148        return total

Token usage and cost for a single LLM call or a multi-message sum.

Carries only the fields that are meaningfully aggregatable across messages: token counts and cost. Per-call latency lives on the individual message's latency_ms field; aggregating it across the full trace would mix latencies from different points in time, so MessageUsage does NOT carry total_llm_latency_ms.

The Usage subclass adds total_llm_latency_ms for the in-flight per-run accumulator that tracks how long this run spent waiting on LLM calls.

input_tokens: int | None

output_tokens: int | None

total_tokens: int | None

cost: float | None

cached_tokens: int | None

@staticmethod

def from_trace( trace: list[typing.Union[openai.types.chat.chat_completion_developer_message_param.ChatCompletionDeveloperMessageParam, openai.types.chat.chat_completion_system_message_param.ChatCompletionSystemMessageParam, openai.types.chat.chat_completion_user_message_param.ChatCompletionUserMessageParam, kiln_ai.utils.open_ai_types.ChatCompletionAssistantMessageParamWrapper, kiln_ai.utils.open_ai_types.ChatCompletionToolMessageParamWrapper, openai.types.chat.chat_completion_function_message_param.ChatCompletionFunctionMessageParam]] | None) -> MessageUsage: View Source

116    @staticmethod
117    def from_trace(
118        trace: "list[ChatCompletionMessageParam] | None",
119    ) -> "MessageUsage":
120        """Sum per-message usage across all assistant messages in a trace.
121
122        Returns MessageUsage() (all fields None) when trace is None/empty
123        or no assistant message has a `usage` field. Skips non-assistant
124        messages and messages where `usage` is missing or None. Always
125        returns a MessageUsage instance — never None.
126
127        Accepts per-message `usage` values that are either MessageUsage
128        instances (including the Usage subclass) or plain dicts (e.g.
129        from JSON round-trips); dicts are validated to MessageUsage
130        before summing.
131        """
132        total: MessageUsage = MessageUsage()
133        if not trace:
134            return total
135
136        for message in trace:
137            if not isinstance(message, dict):
138                continue
139            if message.get("role") != "assistant":
140                continue
141            raw_usage = message.get("usage")
142            if raw_usage is None:
143                continue
144            if isinstance(raw_usage, MessageUsage):
145                total = total + raw_usage
146            elif isinstance(raw_usage, dict):
147                total = total + MessageUsage.model_validate(raw_usage)
148        return total

Sum per-message usage across all assistant messages in a trace.

Returns MessageUsage() (all fields None) when trace is None/empty or no assistant message has a usage field. Skips non-assistant messages and messages where usage is missing or None. Always returns a MessageUsage instance — never None.

Accepts per-message usage values that are either MessageUsage instances (including the Usage subclass) or plain dicts (e.g. from JSON round-trips); dicts are validated to MessageUsage before summing.

model_config: ClassVar[pydantic.config.ConfigDict] = {}

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

class Priority(enum.IntEnum): View Source

 9class Priority(IntEnum):
10    """Priority levels, where P0 is highest priority."""
11
12    p0 = 0
13    p1 = 1
14    p2 = 2
15    p3 = 3

Priority levels, where P0 is highest priority.

p0 = <Priority.p0: 0>

p1 = <Priority.p1: 1>

p2 = <Priority.p2: 2>

p3 = <Priority.p3: 3>

class Prompt(kiln_ai.datamodel.basemodel.KilnParentedModel, kiln_ai.datamodel.BasePrompt): View Source

33class Prompt(KilnParentedModel, BasePrompt):
34    """
35    A prompt for a task. This is the custom prompt parented by a task.
36    """
37
38    pass

A prompt for a task. This is the custom prompt parented by a task.

def relationship_name() -> str: View Source

859        def relationship_name_method() -> str:
860            return filesystem_name

The type of the None singleton.

def parent_type() -> Type[kiln_ai.datamodel.basemodel.KilnParentModel]: View Source

852        def parent_class_method() -> Type[KilnParentModel]:
853            return cls

The type of the None singleton.

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: View Source

365def init_private_attributes(self: BaseModel, context: Any, /) -> None:
366    """This function is meant to behave like a BaseModel method to initialize private attributes.
367
368    It takes context as an argument since that's what pydantic-core passes when calling it.
369
370    Args:
371        self: The BaseModel instance.
372        context: The context.
373    """
374    if getattr(self, '__pydantic_private__', None) is None:
375        pydantic_private = {}
376        for name, private_attr in self.__private_attributes__.items():
377            # Avoid needlessly creating a new dict for the validated data:
378            if private_attr.default_factory_takes_validated_data:
379                default = private_attr.get_default(
380                    call_default_factory=True, validated_data={**self.__dict__, **pydantic_private}
381                )
382            else:
383                default = private_attr.get_default(call_default_factory=True)
384            if default is not PydanticUndefined:
385                pydantic_private[name] = default
386        object_setattr(self, '__pydantic_private__', pydantic_private)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Args: self: The BaseModel instance. context: The context.

class PromptGenerators(builtins.str, enum.Enum): View Source

 9class PromptGenerators(str, Enum):
10    """Built-in prompt generators that can construct a prompt from a task definition."""
11
12    SIMPLE = "simple_prompt_builder"
13    MULTI_SHOT = "multi_shot_prompt_builder"
14    FEW_SHOT = "few_shot_prompt_builder"
15    REPAIRS = "repairs_prompt_builder"
16    SIMPLE_CHAIN_OF_THOUGHT = "simple_chain_of_thought_prompt_builder"
17    FEW_SHOT_CHAIN_OF_THOUGHT = "few_shot_chain_of_thought_prompt_builder"
18    MULTI_SHOT_CHAIN_OF_THOUGHT = "multi_shot_chain_of_thought_prompt_builder"

Built-in prompt generators that can construct a prompt from a task definition.

SIMPLE = <PromptGenerators.SIMPLE: 'simple_prompt_builder'>

MULTI_SHOT = <PromptGenerators.MULTI_SHOT: 'multi_shot_prompt_builder'>

FEW_SHOT = <PromptGenerators.FEW_SHOT: 'few_shot_prompt_builder'>

REPAIRS = <PromptGenerators.REPAIRS: 'repairs_prompt_builder'>

SIMPLE_CHAIN_OF_THOUGHT = <PromptGenerators.SIMPLE_CHAIN_OF_THOUGHT: 'simple_chain_of_thought_prompt_builder'>

FEW_SHOT_CHAIN_OF_THOUGHT = <PromptGenerators.FEW_SHOT_CHAIN_OF_THOUGHT: 'few_shot_chain_of_thought_prompt_builder'>

MULTI_SHOT_CHAIN_OF_THOUGHT = <PromptGenerators.MULTI_SHOT_CHAIN_OF_THOUGHT: 'multi_shot_chain_of_thought_prompt_builder'>

PromptId = typing.Annotated[str, AfterValidator(func=<function <lambda>>)]

class PromptOptimizationJob(kiln_ai.datamodel.basemodel.KilnParentedModel): View Source

12class PromptOptimizationJob(KilnParentedModel):
13    """
14    The Kiln prompt optimization job datamodel.
15    """
16
17    name: FilenameString = Field(description="The name of the prompt optimization job.")
18    description: str | None = Field(
19        default=None,
20        description="A description of the prompt optimization job for you and your team.",
21    )
22    job_id: str = Field(description="The ID of the job on the remote Kiln server.")
23    target_run_config_id: str = Field(
24        description="The ID of the run configuration used for this job."
25    )
26    latest_status: str = Field(
27        default="pending",
28        description="The latest known status of this prompt optimization job (pending, running, succeeded, failed, cancelled). Not updated in real time.",
29    )
30    optimized_prompt: str | None = Field(
31        default=None,
32        description="The optimized prompt result when the job succeeds.",
33    )
34    created_prompt_id: str | None = Field(
35        default=None,
36        description="The ID of the prompt created from this job's result, if any.",
37    )
38    created_run_config_id: str | None = Field(
39        default=None,
40        description="The ID of the run config created from this job's result, if any.",
41    )
42    eval_ids: list[str] = Field(
43        default_factory=list,
44        description="List of eval IDs used for this job.",
45    )
46
47    def parent_task(self) -> "Task | None":
48        """Get the parent task, with proper typing."""
49        if self.parent is None or self.parent.__class__.__name__ != "Task":
50            return None
51        return self.parent  # type: ignore

The Kiln prompt optimization job datamodel.

description: str | None

job_id: str

target_run_config_id: str

latest_status: str

optimized_prompt: str | None

created_prompt_id: str | None

created_run_config_id: str | None

eval_ids: list[str]

def parent_task(self) -> Task | None: View Source

47    def parent_task(self) -> "Task | None":
48        """Get the parent task, with proper typing."""
49        if self.parent is None or self.parent.__class__.__name__ != "Task":
50            return None
51        return self.parent  # type: ignore

Get the parent task, with proper typing.

def relationship_name() -> str: View Source

859        def relationship_name_method() -> str:
860            return filesystem_name

The type of the None singleton.

def parent_type() -> Type[kiln_ai.datamodel.basemodel.KilnParentModel]: View Source

852        def parent_class_method() -> Type[KilnParentModel]:
853            return cls

The type of the None singleton.

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: View Source

365def init_private_attributes(self: BaseModel, context: Any, /) -> None:
366    """This function is meant to behave like a BaseModel method to initialize private attributes.
367
368    It takes context as an argument since that's what pydantic-core passes when calling it.
369
370    Args:
371        self: The BaseModel instance.
372        context: The context.
373    """
374    if getattr(self, '__pydantic_private__', None) is None:
375        pydantic_private = {}
376        for name, private_attr in self.__private_attributes__.items():
377            # Avoid needlessly creating a new dict for the validated data:
378            if private_attr.default_factory_takes_validated_data:
379                default = private_attr.get_default(
380                    call_default_factory=True, validated_data={**self.__dict__, **pydantic_private}
381                )
382            else:
383                default = private_attr.get_default(call_default_factory=True)
384            if default is not PydanticUndefined:
385                pydantic_private[name] = default
386        object_setattr(self, '__pydantic_private__', pydantic_private)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Args: self: The BaseModel instance. context: The context.

class RequirementRating(pydantic.main.BaseModel): View Source

29class RequirementRating(BaseModel):
30    """Rating for a specific requirement within a task output."""
31
32    value: float = Field(
33        description="The rating value. Interpretation depends on rating type"
34    )
35    type: TaskOutputRatingType = Field(description="The type of rating")

Rating for a specific requirement within a task output.

value: float

type: TaskOutputRatingType

model_config: ClassVar[pydantic.config.ConfigDict] = {}

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

class StructuredOutputMode(builtins.str, enum.Enum): View Source

37class StructuredOutputMode(str, Enum):
38    """
39    Enumeration of supported structured output modes.
40
41    - json_schema: request json using API capabilities for json_schema
42    - function_calling: request json using API capabilities for function calling
43    - json_mode: request json using API's JSON mode, which should return valid JSON, but isn't checking/passing the schema
44    - json_instructions: append instructions to the prompt to request json matching the schema. No API capabilities are used. You should have a custom parser on these models as they will be returning strings.
45    - json_instruction_and_object: append instructions to the prompt to request json matching the schema. Also request the response as json_mode via API capabilities (returning dictionaries).
46    - json_custom_instructions: The model should output JSON, but custom instructions are already included in the system prompt. Don't append additional JSON instructions.
47    - default: let the adapter decide (legacy, do not use for new use cases)
48    - unknown: used for cases where the structured output mode is not known (on old models where it wasn't saved). Should lookup best option at runtime.
49    """
50
51    default = "default"
52    json_schema = "json_schema"
53    function_calling_weak = "function_calling_weak"
54    function_calling = "function_calling"
55    json_mode = "json_mode"
56    json_instructions = "json_instructions"
57    json_instruction_and_object = "json_instruction_and_object"
58    json_custom_instructions = "json_custom_instructions"
59    unknown = "unknown"

Enumeration of supported structured output modes.

json_schema: request json using API capabilities for json_schema
function_calling: request json using API capabilities for function calling
json_mode: request json using API's JSON mode, which should return valid JSON, but isn't checking/passing the schema
json_instructions: append instructions to the prompt to request json matching the schema. No API capabilities are used. You should have a custom parser on these models as they will be returning strings.
json_instruction_and_object: append instructions to the prompt to request json matching the schema. Also request the response as json_mode via API capabilities (returning dictionaries).
json_custom_instructions: The model should output JSON, but custom instructions are already included in the system prompt. Don't append additional JSON instructions.
default: let the adapter decide (legacy, do not use for new use cases)
unknown: used for cases where the structured output mode is not known (on old models where it wasn't saved). Should lookup best option at runtime.

default = <StructuredOutputMode.default: 'default'>

json_schema = <StructuredOutputMode.json_schema: 'json_schema'>

function_calling_weak = <StructuredOutputMode.function_calling_weak: 'function_calling_weak'>

function_calling = <StructuredOutputMode.function_calling: 'function_calling'>

json_mode = <StructuredOutputMode.json_mode: 'json_mode'>

json_instructions = <StructuredOutputMode.json_instructions: 'json_instructions'>

json_instruction_and_object = <StructuredOutputMode.json_instruction_and_object: 'json_instruction_and_object'>

json_custom_instructions = <StructuredOutputMode.json_custom_instructions: 'json_custom_instructions'>

unknown = <StructuredOutputMode.unknown: 'unknown'>

class TaskOutput(kiln_ai.datamodel.basemodel.KilnBaseModel): View Source

347class TaskOutput(KilnBaseModel):
348    """
349    An output for a specific task run.
350
351    Contains the actual output content, its source (human or synthetic),
352    and optional rating information.
353    """
354
355    output: str = Field(
356        description="The output of the task. JSON formatted for structured output, plaintext for unstructured output."
357    )
358    source: DataSource | None = Field(
359        description="The source of the output: human or synthetic.",
360        default=None,
361    )
362    rating: TaskOutputRating | None = Field(
363        default=None, description="The rating of the output"
364    )
365
366    def validate_output_format(self, task: "Task") -> Self:
367        # validate output
368        if task.output_json_schema is not None:
369            try:
370                output_parsed = json.loads(self.output)
371            except json.JSONDecodeError:
372                raise ValueError("Output is not a valid JSON object")
373
374            validate_schema_with_value_error(
375                output_parsed,
376                task.output_json_schema,
377                TASK_OUTPUT_SCHEMA_ERROR_PREFIX,
378            )
379        return self
380
381    @model_validator(mode="after")
382    def validate_output_source(self, info: ValidationInfo) -> Self:
383        # On strict mode and not loaded from file, we validate output_source is not None.
384        # We want to be able to load any data, even if it's not perfect. But we want to create perfect data when adding new data.
385        if not strict_mode():
386            return self
387        if self.loaded_from_file(info):
388            return self
389        if self.source is None:
390            raise ValueError("Output source is required when strict mode is enabled")
391        return self

An output for a specific task run.

Contains the actual output content, its source (human or synthetic), and optional rating information.

output: str

source: DataSource | None

rating: TaskOutputRating | None

def validate_output_format(self, task: Task) -> Self: View Source

366    def validate_output_format(self, task: "Task") -> Self:
367        # validate output
368        if task.output_json_schema is not None:
369            try:
370                output_parsed = json.loads(self.output)
371            except json.JSONDecodeError:
372                raise ValueError("Output is not a valid JSON object")
373
374            validate_schema_with_value_error(
375                output_parsed,
376                task.output_json_schema,
377                TASK_OUTPUT_SCHEMA_ERROR_PREFIX,
378            )
379        return self

@model_validator(mode='after')

def validate_output_source(self, info: pydantic_core.core_schema.ValidationInfo) -> Self: View Source

381    @model_validator(mode="after")
382    def validate_output_source(self, info: ValidationInfo) -> Self:
383        # On strict mode and not loaded from file, we validate output_source is not None.
384        # We want to be able to load any data, even if it's not perfect. But we want to create perfect data when adding new data.
385        if not strict_mode():
386            return self
387        if self.loaded_from_file(info):
388            return self
389        if self.source is None:
390            raise ValueError("Output source is required when strict mode is enabled")
391        return self

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: View Source

365def init_private_attributes(self: BaseModel, context: Any, /) -> None:
366    """This function is meant to behave like a BaseModel method to initialize private attributes.
367
368    It takes context as an argument since that's what pydantic-core passes when calling it.
369
370    Args:
371        self: The BaseModel instance.
372        context: The context.
373    """
374    if getattr(self, '__pydantic_private__', None) is None:
375        pydantic_private = {}
376        for name, private_attr in self.__private_attributes__.items():
377            # Avoid needlessly creating a new dict for the validated data:
378            if private_attr.default_factory_takes_validated_data:
379                default = private_attr.get_default(
380                    call_default_factory=True, validated_data={**self.__dict__, **pydantic_private}
381                )
382            else:
383                default = private_attr.get_default(call_default_factory=True)
384            if default is not PydanticUndefined:
385                pydantic_private[name] = default
386        object_setattr(self, '__pydantic_private__', pydantic_private)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Args: self: The BaseModel instance. context: The context.

class TaskOutputRating(kiln_ai.datamodel.basemodel.KilnBaseModel): View Source

 59class TaskOutputRating(KilnBaseModel):
 60    """
 61    A rating for a task output, including an overall rating and ratings for each requirement.
 62
 63    Supports:
 64    - five_star: 1-5 star ratings
 65    - pass_fail: boolean pass/fail (1.0 = pass, 0.0 = fail)
 66    - pass_fail_critical: tri-state (1.0 = pass, 0.0 = fail, -1.0 = critical fail)
 67    """
 68
 69    type: TaskOutputRatingType = Field(
 70        default=TaskOutputRatingType.five_star,
 71        description="The rating system used for this rating.",
 72    )
 73    value: float | None = Field(
 74        description="The rating value. Interpretation depends on rating type:\n- five_star: 1-5 stars\n- pass_fail: 1.0 (pass) or 0.0 (fail)\n- pass_fail_critical: 1.0 (pass), 0.0 (fail), or -1.0 (critical fail)",
 75        default=None,
 76    )
 77    requirement_ratings: Dict[ID_TYPE, RequirementRating] = Field(
 78        default={},
 79        description="The ratings of the requirements of the task. The ID can be either a task_requirement_id or a named rating for an eval_output_score name (in format 'named::<name>').",
 80    )
 81
 82    # Previously we stored rating values as a dict of floats, but now we store them as RequirementRating objects.
 83    @model_validator(mode="before")
 84    def upgrade_old_format(cls, data: dict) -> dict:
 85        if not isinstance(data, dict):
 86            return data
 87
 88        # Check if we have the old format (dict of floats)
 89        req_ratings = data.get("requirement_ratings", {})
 90        if req_ratings and all(
 91            isinstance(v, (int, float)) for v in req_ratings.values()
 92        ):
 93            # Convert each float to a RequirementRating object
 94            # all ratings are five star at the point we used this format
 95            data["requirement_ratings"] = {
 96                k: {"value": v, "type": TaskOutputRatingType.five_star}
 97                for k, v in req_ratings.items()
 98            }
 99
100        return data
101
102    # Used to select high quality outputs for example selection (MultiShotPromptBuilder, etc)
103    def is_high_quality(self) -> bool:
104        if self.value is None:
105            return False
106
107        if self.type == TaskOutputRatingType.five_star:
108            return self.value >= 4
109        elif self.type == TaskOutputRatingType.pass_fail:
110            return self.value == 1.0
111        elif self.type == TaskOutputRatingType.pass_fail_critical:
112            return self.value == 1.0
113        return False
114
115    @model_validator(mode="after")
116    def validate_rating(self) -> Self:
117        if self.type not in TaskOutputRatingType:
118            raise ValueError(f"Invalid rating type: {self.type}")
119
120        # Overall rating is optional
121        if self.value is not None:
122            self._validate_rating(self.type, self.value, "overall rating")
123
124        for req_id, req_rating in self.requirement_ratings.items():
125            self._validate_rating(
126                req_rating.type,
127                req_rating.value,
128                f"requirement rating for req ID: {req_id}",
129            )
130
131        return self
132
133    def _validate_rating(
134        self, type: TaskOutputRatingType, rating: float | None, rating_name: str
135    ) -> None:
136        if type == TaskOutputRatingType.five_star:
137            self._validate_five_star(rating, rating_name)
138        elif type == TaskOutputRatingType.pass_fail:
139            self._validate_pass_fail(rating, rating_name)
140        elif type == TaskOutputRatingType.pass_fail_critical:
141            self._validate_pass_fail_critical(rating, rating_name)
142
143    def _validate_five_star(self, rating: float | None, rating_name: str) -> None:
144        if rating is None or not isinstance(rating, float) or not rating.is_integer():
145            raise ValueError(
146                f"{rating_name.capitalize()} of type five_star must be an integer value (1-5)"
147            )
148        if rating < 1 or rating > 5:
149            raise ValueError(
150                f"{rating_name.capitalize()} of type five_star must be between 1 and 5 stars"
151            )
152
153    def _validate_pass_fail(self, rating: float | None, rating_name: str) -> None:
154        if rating is None or not isinstance(rating, float) or not rating.is_integer():
155            raise ValueError(
156                f"{rating_name.capitalize()} of type pass_fail must be an integer value (0 or 1)"
157            )
158        if rating not in [0, 1]:
159            raise ValueError(
160                f"{rating_name.capitalize()} of type pass_fail must be 0 (fail) or 1 (pass)"
161            )
162
163    def _validate_pass_fail_critical(
164        self, rating: float | None, rating_name: str
165    ) -> None:
166        if rating is None or not isinstance(rating, float) or not rating.is_integer():
167            raise ValueError(
168                f"{rating_name.capitalize()} of type pass_fail_critical must be an integer value (-1, 0, or 1)"
169            )
170        if rating not in [-1, 0, 1]:
171            raise ValueError(
172                f"{rating_name.capitalize()} of type pass_fail_critical must be -1 (critical fail), 0 (fail), or 1 (pass)"
173            )

A rating for a task output, including an overall rating and ratings for each requirement.

Supports:

five_star: 1-5 star ratings
pass_fail: boolean pass/fail (1.0 = pass, 0.0 = fail)
pass_fail_critical: tri-state (1.0 = pass, 0.0 = fail, -1.0 = critical fail)

type: TaskOutputRatingType

value: float | None

requirement_ratings: Dict[Optional[str], RequirementRating]

@model_validator(mode='before')

def upgrade_old_format(cls, data: dict) -> dict: View Source

 83    @model_validator(mode="before")
 84    def upgrade_old_format(cls, data: dict) -> dict:
 85        if not isinstance(data, dict):
 86            return data
 87
 88        # Check if we have the old format (dict of floats)
 89        req_ratings = data.get("requirement_ratings", {})
 90        if req_ratings and all(
 91            isinstance(v, (int, float)) for v in req_ratings.values()
 92        ):
 93            # Convert each float to a RequirementRating object
 94            # all ratings are five star at the point we used this format
 95            data["requirement_ratings"] = {
 96                k: {"value": v, "type": TaskOutputRatingType.five_star}
 97                for k, v in req_ratings.items()
 98            }
 99
100        return data

def is_high_quality(self) -> bool: View Source

103    def is_high_quality(self) -> bool:
104        if self.value is None:
105            return False
106
107        if self.type == TaskOutputRatingType.five_star:
108            return self.value >= 4
109        elif self.type == TaskOutputRatingType.pass_fail:
110            return self.value == 1.0
111        elif self.type == TaskOutputRatingType.pass_fail_critical:
112            return self.value == 1.0
113        return False

@model_validator(mode='after')

def validate_rating(self) -> Self: View Source

115    @model_validator(mode="after")
116    def validate_rating(self) -> Self:
117        if self.type not in TaskOutputRatingType:
118            raise ValueError(f"Invalid rating type: {self.type}")
119
120        # Overall rating is optional
121        if self.value is not None:
122            self._validate_rating(self.type, self.value, "overall rating")
123
124        for req_id, req_rating in self.requirement_ratings.items():
125            self._validate_rating(
126                req_rating.type,
127                req_rating.value,
128                f"requirement rating for req ID: {req_id}",
129            )
130
131        return self

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: View Source

365def init_private_attributes(self: BaseModel, context: Any, /) -> None:
366    """This function is meant to behave like a BaseModel method to initialize private attributes.
367
368    It takes context as an argument since that's what pydantic-core passes when calling it.
369
370    Args:
371        self: The BaseModel instance.
372        context: The context.
373    """
374    if getattr(self, '__pydantic_private__', None) is None:
375        pydantic_private = {}
376        for name, private_attr in self.__private_attributes__.items():
377            # Avoid needlessly creating a new dict for the validated data:
378            if private_attr.default_factory_takes_validated_data:
379                default = private_attr.get_default(
380                    call_default_factory=True, validated_data={**self.__dict__, **pydantic_private}
381                )
382            else:
383                default = private_attr.get_default(call_default_factory=True)
384            if default is not PydanticUndefined:
385                pydantic_private[name] = default
386        object_setattr(self, '__pydantic_private__', pydantic_private)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Args: self: The BaseModel instance. context: The context.

class TaskOutputRatingType(builtins.str, enum.Enum): View Source

28class TaskOutputRatingType(str, Enum):
29    """Defines the types of rating systems available for task outputs."""
30
31    five_star = "five_star"
32    pass_fail = "pass_fail"
33    pass_fail_critical = "pass_fail_critical"
34    custom = "custom"

Defines the types of rating systems available for task outputs.

five_star = <TaskOutputRatingType.five_star: 'five_star'>

pass_fail = <TaskOutputRatingType.pass_fail: 'pass_fail'>

pass_fail_critical = <TaskOutputRatingType.pass_fail_critical: 'pass_fail_critical'>

custom = <TaskOutputRatingType.custom: 'custom'>

class TaskRequirement(pydantic.main.BaseModel): View Source

40class TaskRequirement(BaseModel):
41    """
42    Defines a specific requirement that should be met by task outputs.
43
44    Includes an identifier, name, description, instruction for meeting the requirement,
45    priority level, and rating type (five_star, pass_fail, pass_fail_critical, custom).
46    """
47
48    id: ID_TYPE = ID_FIELD
49    name: FilenameStringShort = Field(description="The name of the task requirement.")
50    description: str | None = Field(
51        default=None,
52        description="Optional elaboration on the requirement's purpose.",
53    )
54    instruction: str = Field(
55        min_length=1, description="Instructions for meeting the requirement."
56    )
57    priority: Priority = Field(
58        default=Priority.p2, description="The priority level of the requirement."
59    )
60    type: TaskOutputRatingType = Field(
61        default=TaskOutputRatingType.five_star,
62        description="The rating type used to evaluate this requirement.",
63    )

Defines a specific requirement that should be met by task outputs.

Includes an identifier, name, description, instruction for meeting the requirement, priority level, and rating type (five_star, pass_fail, pass_fail_critical, custom).

id: Optional[str]

name: Annotated[str, BeforeValidator(func=<function name_validator.<locals>.fn at 0x7f5eb7c1ade0>, json_schema_input_type=PydanticUndefined), StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=32, pattern=None, ascii_only=None)]

description: str | None

instruction: str

priority: Priority

type: TaskOutputRatingType

model_config: ClassVar[pydantic.config.ConfigDict] = {}

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

class Usage(kiln_ai.datamodel.MessageUsage): View Source

151class Usage(MessageUsage):
152    """Token usage, cost, and aggregate LLM latency for a per-run accumulator.
153
154    Extends :class:`MessageUsage` with ``total_llm_latency_ms``, which is
155    only meaningful while a single run is in flight (its model calls run
156    sequentially in real time). For per-message records and full-trace
157    sums use :class:`MessageUsage` — those values would mix latencies
158    from different points in time, so the field doesn't apply.
159    """
160
161    total_llm_latency_ms: int | None = Field(
162        default=None,
163        description="Total time spent waiting on LLM API calls in milliseconds. Sum of per-call latencies, excludes tool execution time.",
164        ge=0,
165    )
166
167    def __add__(self, other: "MessageUsage | Usage") -> "Usage":
168        """Add Usage to either Usage or MessageUsage.
169
170        Token / cost fields sum the same way as :meth:`MessageUsage.__add__`.
171
172        ``Usage + Usage`` sums both ``total_llm_latency_ms`` values
173        (None-graceful). ``Usage + MessageUsage`` carries ``self``'s
174        ``total_llm_latency_ms`` through unchanged — the right-hand side
175        has no latency to contribute.
176
177        Always returns a :class:`Usage` so chained ``usage += msg_usage``
178        keeps the latency on the accumulator.
179        """
180        if not isinstance(other, MessageUsage):
181            raise TypeError(f"Cannot add Usage with {type(other).__name__}")
182
183        other_latency = other.total_llm_latency_ms if isinstance(other, Usage) else None
184
185        return Usage(
186            input_tokens=_add_optional_int(self.input_tokens, other.input_tokens),
187            output_tokens=_add_optional_int(self.output_tokens, other.output_tokens),
188            total_tokens=_add_optional_int(self.total_tokens, other.total_tokens),
189            cost=_add_optional_float(self.cost, other.cost),
190            cached_tokens=_add_optional_int(self.cached_tokens, other.cached_tokens),
191            total_llm_latency_ms=_add_optional_int(
192                self.total_llm_latency_ms, other_latency
193            ),
194        )

Token usage, cost, and aggregate LLM latency for a per-run accumulator.

Extends MessageUsage with total_llm_latency_ms, which is only meaningful while a single run is in flight (its model calls run sequentially in real time). For per-message records and full-trace sums use MessageUsage — those values would mix latencies from different points in time, so the field doesn't apply.

total_llm_latency_ms: int | None

model_config: ClassVar[pydantic.config.ConfigDict] = {}

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

def generate_model_id() -> str: View Source

38def generate_model_id() -> str:
39    return str(uuid.uuid4().int)[:12]

prompt_generator_values = ['simple_prompt_builder', 'multi_shot_prompt_builder', 'few_shot_prompt_builder', 'repairs_prompt_builder', 'simple_chain_of_thought_prompt_builder', 'few_shot_chain_of_thought_prompt_builder', 'multi_shot_chain_of_thought_prompt_builder']