Typed output¶
So far triage returns a line of text that your code would have to parse. Give the task Pydantic types instead, and the harness asks the model for structured output, parses the answer, and checks it before your code sees it.
Declare the types¶
Create a package for your app's code, helpdesk/, with an empty __init__.py and this models.py:
"""What goes into the support desk's tasks and what comes out."""
from typing import Literal
from pydantic import BaseModel, Field
class Ticket(BaseModel):
"""A message a customer sent to the support desk."""
text: str
class Triage(BaseModel):
"""How the desk should handle a ticket."""
category: Literal["billing", "bug", "account", "other"]
priority: int = Field(ge=1, le=4, description="1 is the most urgent.")
summary: str = Field(description="One sentence: what the customer wants.")
These are ordinary Pydantic models. The Field constraints and descriptions are part of the contract: the model sees the descriptions, and the harness enforces the constraints.
Use them in the task¶
apiVersion: harnessweaver/v1
kind: Harness
metadata:
name: support-desk
owner: support-team
models:
provider: openai.chat
base_url: http://localhost:4000
api_key_env: LITELLM_API_KEY
default: fast
aliases: { fast: fast-chat, strong: strong-chat }
prices:
fast-chat: { input_per_mtok: 0.25, output_per_mtok: 2.00 }
strong-chat: { input_per_mtok: 3.00, output_per_mtok: 15.00 }
tasks:
triage:
runner: direct
model: fast
prompt: triage@2
input: helpdesk.models:Ticket
output: helpdesk.models:Triage
inputis the type the task's input is validated into, aspackage.module:Name.outputis the type the model's answer is parsed into.strmeans plain text.
The prompt no longer needs to describe a reply format, so it gets a new version:
You triage messages sent to Acme's support desk.
Decide the category (billing, bug, account or other), the priority (1 is the
most urgent, 4 the least), and write a one-sentence summary of what the
customer wants.
Run it¶
The offline model now answers with structured output:
models:
provider: scripted
replies:
- output:
category: billing
priority: 2
summary: Charged twice for the Pro plan and wants one payment refunded.
import asyncio
from helpdesk.models import Ticket, Triage
import harnessweaver as hw
async def main() -> None:
harness = hw.load("harness.yaml")
ticket = Ticket(text="I was charged twice for my Pro plan. Refund one?")
result = await harness.run("triage", ticket)
print(result.status.value)
if result.ok:
triage = Triage.model_validate(result.output)
print(f"{triage.category}, priority {triage.priority}: {triage.summary}")
if __name__ == "__main__":
asyncio.run(main())
$ HARNESSWEAVER_ENV=offline uv run python main.py
completed
billing, priority 2: Charged twice for the Pro plan and wants one payment refunded.
result.output is the parsed output as JSON (a dict here), because a RunResult travels between processes and languages. Triage.model_validate(result.output) turns it back into your type.
What the harness did¶
- It validated the input into a
Ticket. Input that doesn't validate ends the runfailed(codevalidation_failed) before the model is called. - It sent
Triage's JSON Schema with the request, in the form the provider's API expects (for example, OpenAI'sresponse_format). - It parsed the answer into
Triage: the category must be one of the four, and the priority between 1 and 4. - It ran the
output.beforehook on the parsed output, where output guards and business rules go.
An answer that doesn't parse is not your code's problem: the next chapter shows the harness asking the model to fix it.
Check
Declaring an output type for every task is one of the requirements of adoption level L1 (Adoption levels). Typed output is where governance starts: you can only check what you can name.
Recap¶
- A task's
inputandoutputname Pydantic models, aspackage.module:Name. - The harness asks for structured output, parses the answer into your type, and checks it on
output.before. result.outputis JSON. Validate it back into your model withmodel_validate.