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.code_tool import CodeTool
 26from kiln_ai.datamodel.data_guide import DataGuide
 27from kiln_ai.datamodel.datamodel_enums import (
 28    FeedbackSource,
 29    FineTuneStatusType,
 30    Priority,
 31    StructuredOutputMode,
 32    TaskOutputRatingType,
 33)
 34from kiln_ai.datamodel.dataset_split import DatasetSplit, DatasetSplitDefinition
 35from kiln_ai.datamodel.external_tool_server import ExternalToolServer
 36from kiln_ai.datamodel.feedback import Feedback
 37from kiln_ai.datamodel.finetune import Finetune
 38from kiln_ai.datamodel.project import Project
 39from kiln_ai.datamodel.prompt import BasePrompt, Prompt
 40from kiln_ai.datamodel.prompt_id import (
 41    PromptGenerators,
 42    PromptId,
 43    prompt_generator_values,
 44)
 45from kiln_ai.datamodel.prompt_optimization_job import PromptOptimizationJob
 46from kiln_ai.datamodel.skill import Skill
 47from kiln_ai.datamodel.task import Task, TaskRequirement
 48from kiln_ai.datamodel.task_output import (
 49    DataSource,
 50    DataSourceProperty,
 51    DataSourceType,
 52    RequirementRating,
 53    TaskOutput,
 54    TaskOutputRating,
 55)
 56from kiln_ai.datamodel.task_run import (
 57    EvalItemSource,
 58    MessageUsage,
 59    TaskRun,
 60    Usage,
 61)
 62
 63__all__ = [
 64    "BasePrompt",
 65    "CodeTool",
 66    "DataGuide",
 67    "DataSource",
 68    "DataSourceProperty",
 69    "DataSourceType",
 70    "DatasetSplit",
 71    "DatasetSplitDefinition",
 72    "EvalItemSource",
 73    "ExternalToolServer",
 74    "Feedback",
 75    "FeedbackSource",
 76    "FineTuneStatusType",
 77    "Finetune",
 78    "MessageUsage",
 79    "Priority",
 80    "Project",
 81    "Prompt",
 82    "PromptGenerators",
 83    "PromptId",
 84    "PromptOptimizationJob",
 85    "RequirementRating",
 86    "Skill",
 87    "StructuredOutputMode",
 88    "Task",
 89    "TaskOutput",
 90    "TaskOutputRating",
 91    "TaskOutputRatingType",
 92    "TaskRequirement",
 93    "TaskRun",
 94    "Usage",
 95    "chunk",
 96    "dataset_split",
 97    "embedding",
 98    "eval",
 99    "extraction",
100    "generate_model_id",
101    "prompt_generator_values",
102    "rag",
103    "reranker",
104    "strict_mode",
105]

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 0x7fc4f95a07c0>, 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 CodeTool(kiln_ai.datamodel.basemodel.KilnParentedModel): View Source

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

@field_validator('tool_function_name')

@classmethod

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

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

@model_validator(mode='after')

def validate_parameters_schema(self) -> Self: View Source

139    @model_validator(mode="after")
140    def validate_parameters_schema(self) -> Self:
141        validate_schema_dict(self.parameters_schema, require_object=True)
142        return self

@model_validator(mode='after')

def validate_code(self) -> Self: View Source

144    @model_validator(mode="after")
145    def validate_code(self) -> Self:
146        code_bytes = self.code.encode("utf-8")
147        if len(code_bytes) > 64 * 1024:
148            raise ValueError(
149                f"Code is too large ({len(code_bytes)} bytes). Maximum size is 64KB."
150            )
151
152        try:
153            compile(self.code, "<code_tool>", "exec")
154        except SyntaxError as e:
155            raise ValueError(f"Code has a syntax error: {e}") from e
156
157        tree = ast.parse(self.code)
158        has_run_fn = any(
159            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
160            and node.name == "run"
161            for node in ast.iter_child_nodes(tree)
162        )
163        if not has_run_fn:
164            raise ValueError(
165                "Code must define a module-level 'run' function (def run(...) or async def run(...))."
166            )
167
168        return self

@model_validator(mode='after')

def validate_allowlist(self) -> Self: View Source

170    @model_validator(mode="after")
171    def validate_allowlist(self) -> Self:
172        validate_tool_allowlist(
173            self.tool_allowlist,
174            caller="code tools",
175            self_tool_id=build_code_tool_id(self.id) if self.id is not None else None,
176        )
177        return self

def relationship_name() -> str: View Source

856        def relationship_name_method() -> str:
857            return filesystem_name

The type of the None singleton.

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

849        def parent_class_method() -> Type[KilnParentModel]:
850            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, matching Mike's
30      GENERATE_CORPUS_GUIDELINES vocabulary.
31
32    The metaprompter treats the whole body as one editable artifact and returns
33    a refined version on each refine pass; refine auto-detects which shape it
34    is working on by checking for a `# Reference Inputs` heading.
35    """
36
37    guide: str = Field(
38        default="",
39        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`.",
40    )
41
42    source: DataGuideSource = Field(
43        default="manual",
44        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).",
45    )

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, matching Mike's GENERATE_CORPUS_GUIDELINES vocabulary.

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

856        def relationship_name_method() -> str:
857            return filesystem_name

The type of the None singleton.

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

849        def parent_class_method() -> Type[KilnParentModel]:
850            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

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

218    @model_validator(mode="after")
219    def normalize_empty_run_config_id(self) -> Self:
220        # Some callers (e.g. tool wrappers reading from external properties)
221        # default missing IDs to "". Coerce to None so this field is either a
222        # real ID or unset — never an ambiguous empty string on disk.
223        if self.run_config_id == "":
224            self.run_config_id = None
225        return self

@model_validator(mode='after')

def validate_type(self) -> DataSource: View Source

300    @model_validator(mode="after")
301    def validate_type(self) -> "DataSource":
302        if self.type not in DataSourceType:
303            raise ValueError(f"Invalid data source type: {self.type}")
304        return self

@model_validator(mode='after')

def validate_properties(self) -> DataSource: View Source

306    @model_validator(mode="after")
307    def validate_properties(self) -> "DataSource":
308        for prop in self._data_source_properties:
309            # Check the property type is correct
310            if prop.name in self.properties:
311                if not isinstance(self.properties[prop.name], prop.type):
312                    raise ValueError(
313                        f"'{prop.name}' must be of type {prop.type.__name__} for {self.type} data source"
314                    )
315            # Check the property is required for the data source type
316            if self.type in prop.required_for:
317                if prop.name not in self.properties:
318                    raise ValueError(
319                        f"'{prop.name}' is required for {self.type} data source"
320                    )
321            # Check the property is not allowed for the data source type
322            elif self.type in prop.not_allowed_for and prop.name in self.properties:
323                raise ValueError(
324                    f"'{prop.name}' is not allowed for {self.type} data source"
325                )
326        return self

@model_validator(mode='after')

def validate_no_empty_properties(self) -> Self: View Source

328    @model_validator(mode="after")
329    def validate_no_empty_properties(self) -> Self:
330        for prop, value in self.properties.items():
331            if isinstance(value, str) and value == "":
332                raise ValueError(
333                    f"Property '{prop}' must be a non-empty string for {self.type} data source"
334                )
335        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

181class DataSourceProperty(BaseModel):
182    """
183    Defines a property that can be associated with a data source.
184
185    Includes validation rules for when properties are required or not allowed
186    based on the data source type.
187    """
188
189    name: str
190    type: Type[Union[str, int, float]]
191    required_for: List[DataSourceType] = []
192    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

167class DataSourceType(str, Enum):
168    """
169    The source type of a piece of data.
170
171    Human: a human created the data
172    Synthetic: a model created the data
173    """
174
175    human = "human"
176    synthetic = "synthetic"
177    file_import = "file_import"
178    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

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

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

856        def relationship_name_method() -> str:
857            return filesystem_name

The type of the None singleton.

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

849        def parent_class_method() -> Type[KilnParentModel]:
850            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

 94class FeedbackSource(str, Enum):
 95    """Where a piece of feedback originated.
 96
 97    This is an append-only enum: new sources can be added freely, but existing
 98    values must never be removed or renamed so that older persisted data
 99    continues to load.
100    """
101
102    run_page = "run-page"
103    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

856        def relationship_name_method() -> str:
857            return filesystem_name

The type of the None singleton.

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

849        def parent_class_method() -> Type[KilnParentModel]:
850            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 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

856        def relationship_name_method() -> str:
857            return filesystem_name

The type of the None singleton.

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

849        def parent_class_method() -> Type[KilnParentModel]:
850            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

856        def relationship_name_method() -> str:
857            return filesystem_name

The type of the None singleton.

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

849        def parent_class_method() -> Type[KilnParentModel]:
850            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

20class RequirementRating(BaseModel):
21    """Rating for a specific requirement within a task output."""
22
23    value: float = Field(
24        description="The rating value. Interpretation depends on rating type"
25    )
26    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

338class TaskOutput(KilnBaseModel):
339    """
340    An output for a specific task run.
341
342    Contains the actual output content, its source (human or synthetic),
343    and optional rating information.
344    """
345
346    output: str = Field(
347        description="The output of the task. JSON formatted for structured output, plaintext for unstructured output."
348    )
349    source: DataSource | None = Field(
350        description="The source of the output: human or synthetic.",
351        default=None,
352    )
353    rating: TaskOutputRating | None = Field(
354        default=None, description="The rating of the output"
355    )
356
357    def validate_output_format(self, task: "Task") -> Self:
358        # validate output
359        if task.output_json_schema is not None:
360            try:
361                output_parsed = json.loads(self.output)
362            except json.JSONDecodeError:
363                raise ValueError("Output is not a valid JSON object")
364
365            validate_schema_with_value_error(
366                output_parsed,
367                task.output_json_schema,
368                "This task requires a specific output schema. While the model produced JSON, that JSON didn't meet the schema. Search 'Troubleshooting Structured Data Issues' in our docs for more information.",
369            )
370        return self
371
372    @model_validator(mode="after")
373    def validate_output_source(self, info: ValidationInfo) -> Self:
374        # On strict mode and not loaded from file, we validate output_source is not None.
375        # 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.
376        if not strict_mode():
377            return self
378        if self.loaded_from_file(info):
379            return self
380        if self.source is None:
381            raise ValueError("Output source is required when strict mode is enabled")
382        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

357    def validate_output_format(self, task: "Task") -> Self:
358        # validate output
359        if task.output_json_schema is not None:
360            try:
361                output_parsed = json.loads(self.output)
362            except json.JSONDecodeError:
363                raise ValueError("Output is not a valid JSON object")
364
365            validate_schema_with_value_error(
366                output_parsed,
367                task.output_json_schema,
368                "This task requires a specific output schema. While the model produced JSON, that JSON didn't meet the schema. Search 'Troubleshooting Structured Data Issues' in our docs for more information.",
369            )
370        return self

@model_validator(mode='after')

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

372    @model_validator(mode="after")
373    def validate_output_source(self, info: ValidationInfo) -> Self:
374        # On strict mode and not loaded from file, we validate output_source is not None.
375        # 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.
376        if not strict_mode():
377            return self
378        if self.loaded_from_file(info):
379            return self
380        if self.source is None:
381            raise ValueError("Output source is required when strict mode is enabled")
382        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

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

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

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

def is_high_quality(self) -> bool: View Source

 94    def is_high_quality(self) -> bool:
 95        if self.value is None:
 96            return False
 97
 98        if self.type == TaskOutputRatingType.five_star:
 99            return self.value >= 4
100        elif self.type == TaskOutputRatingType.pass_fail:
101            return self.value == 1.0
102        elif self.type == TaskOutputRatingType.pass_fail_critical:
103            return self.value == 1.0
104        return False

@model_validator(mode='after')

def validate_rating(self) -> Self: View Source

106    @model_validator(mode="after")
107    def validate_rating(self) -> Self:
108        if self.type not in TaskOutputRatingType:
109            raise ValueError(f"Invalid rating type: {self.type}")
110
111        # Overall rating is optional
112        if self.value is not None:
113            self._validate_rating(self.type, self.value, "overall rating")
114
115        for req_id, req_rating in self.requirement_ratings.items():
116            self._validate_rating(
117                req_rating.type,
118                req_rating.value,
119                f"requirement rating for req ID: {req_id}",
120            )
121
122        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

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

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 0x7fc4f95a1760>, 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']