Insurance Claims Triage Agents with Structured Classification

First notice of loss is where an insurance claim is won or lost: the wrong severity bucket delays a roof repair by three weeks, and a missed fraud signal pays out money nobody will ask for again. The instinct with an LLM is to let it write a paragraph assessment and have a human skim it. Resist that. What you want at intake is a machine that emits a typed decision: a claim category, a severity, coverage flags, fraud signals, a confidence value, and citations back to the policy documents it relied on, every single time, validated before it leaves the agent.

The triage schema is the contract

In Pydantic AI, output_type does double duty: the model builds a JSON schema from your Pydantic model so the LLM knows the shape to fill, and the same model validates the data at the end of the run, prompting the agent to try again if validation fails. So the schema is simultaneously your prompt, your type checker, and your runtime assertion.

from typing import Literal

from pydantic import BaseModel, Field


class Evidence(BaseModel):
    """A verbatim anchor into a policy document. No paraphrase allowed."""
    document_id: str = Field(description='Policy or endorsement identifier, e.g. HO-2024-8891')
    section: str = Field(description='Clause reference, e.g. Section IV(3)(b)')
    quote: str = Field(description='Exact sentence copied from the retrieved clause')


class FraudSignal(BaseModel):
    code: Literal[
        'late_reporting', 'prior_claims_proximity', 'inconsistent_statement',
        'coverage_recently_added', 'total_loss_of_uninsured_item',
    ]
    reason: str
    evidence: list[Evidence] = Field(min_length=1)


class CoverageFlags(BaseModel):
    policy_in_force: bool
    deductible_usd: int = Field(ge=0)
    applicable_limit_usd: int = Field(ge=0)
    exclusions_matched: list[str] = Field(default_factory=list)
    awaiting_human_confirmation: list[str] = Field(
        default_factory=list,
        description='Facts the agent could not confirm from documents',
    )


class TriageDecision(BaseModel):
    claim_type: Literal[
        'auto_collision', 'property_damage', 'theft', 'liability',
        'weather', 'travel', 'health',
    ]
    severity: Literal['low', 'medium', 'high', 'catastrophic']
    coverage: CoverageFlags
    fraud_signals: list[FraudSignal] = Field(default_factory=list)
    confidence: float = Field(ge=0, le=1, description='Calibrated confidence in claim_type')
    evidence: list[Evidence] = Field(min_length=1)
    summary: str = Field(description='One sentence, no numbers the documents do not contain')
    policy_version: str

Notice what the schema forbids by construction. min_length=1 on evidence means an uncited decision is not a valid decision, so it cannot validate. awaiting_human_confirmation is a first-class field for facts the documents do not settle, which gives the model an honest place to put “I don’t know” instead of inventing a deductible. And field descriptions are sent to the model as part of the schema, so write them as instructions rather than labels.

Wire it up:

from dataclasses import dataclass

from pydantic_ai import Agent, Capability, RunContext


@dataclass
class TriageDeps:
    tenant_id: str
    policy_store: PolicyStore
    corpus_version: str


policy_corpus = Capability[TriageDeps](
    id='policy-corpus',
    description='Verbatim policy and endorsement clauses. The only source of policy facts.',
)


triage_agent = Agent(
    'openai:gpt-6-sol',
    deps_type=TriageDeps,
    output_type=TriageDecision,
    instructions=(
        'You classify incoming insurance claims at first notice of loss. '
        'You may only state policy facts you retrieved from a tool in this run. '
        'Never quote a policy from memory. If a fact is not in the retrieved '
        'text, list it under awaiting_human_confirmation. '
        'Set confidence low whenever the evidence is thin.'
    ),
    capabilities=[policy_corpus],
)

Ground every determination in policy documents

Hallucinated policy facts are the failure mode that ends careers, and the fix is mechanical rather than clever: make retrieval a tool, require the citation, then check the quote against what the tool actually returned.

from pydantic_ai import RunContext, ModelRetry


@policy_corpus.tool
async def get_policy_clause(
    ctx: RunContext[TriageDeps], document_id: str, section: str
) -> str:
    """Fetch the verbatim text of a policy clause. The only source of policy facts."""
    return await ctx.deps.policy_store.fetch(document_id, section)


@triage_agent.output_validator
def evidence_must_be_real(ctx: RunContext, out: TriageDecision) -> TriageDecision:
    for item in [*out.evidence, *[s.evidence for s in out.fraud_signals]]:
        source = ctx.deps.policy_store.lookup(item.document_id, item.section)
        if source is None:
            raise ModelRetry(f'Unknown document/section: {item.document_id} {item.section}')
        if item.quote not in source:
            raise ModelRetry(
                f'The quote for {item.section} is not verbatim in the retrieved text. '
                'Copy it exactly or drop the claim.'
            )
    if out.policy_version != ctx.deps.corpus_version:
        raise ModelRetry(
            f'policy_version {out.policy_version} does not match the current corpus '
            f'{ctx.deps.corpus_version}; re-fetch and cite the current text.'
        )
    return out

That substring check converts “did the model invent a clause” from an argument into a unit test. On top of it, add an output guardrail that refuses to expose an answer without citations, using the retry outcome rather than block so the model gets one more pass:

from pydantic_ai_harness import GuardrailResult, OutputGuardrail


def must_cite_sources(output: object) -> GuardrailResult:
    if not getattr(output, 'evidence', None):
        return GuardrailResult.retry('Include at least one source citation.')
    if getattr(output, 'confidence', 1.0) < 0.5:
        return GuardrailResult.retry(
            'Confidence is below the auto-triage floor. State what is missing '
            'and route it for human review instead.'
        )
    return GuardrailResult.allow()


# Attach it on the agent defined above:
#   capabilities=[policy_corpus, OutputGuardrail(guard=must_cite_sources)]

Routing rules by class belong in code

Let the model classify. Do not let it decide the queue. Routing is a pure function over the structured output, which makes it testable, versioned, and explainable to a regulator:

QUEUE = {
    ('auto_collision', 'low'): 'fast_track',
    ('auto_collision', 'catastrophic'): 'major_loss_unit',
    ('property_damage', 'medium'): 'property_desk',
    ('theft', 'medium'): 'siu_review',          # special investigations unit
    ('liability', 'high'): 'legal_liaison',
}

AUTO_REVIEW_TRIGGERS = {
    'fraud_signal': lambda d: bool(d.fraud_signals),
    'low_confidence': lambda d: d.confidence < 0.75,
    'unconfirmed_coverage': lambda d: d.coverage.awaiting_human_confirmation,
    'exclusion_matched': lambda d: bool(d.coverage.exclusions_matched),
    'floor_severity': lambda d: d.severity == 'catastrophic',
}


def route(decision: TriageDecision) -> Routing:
    hits = [name for name, test in AUTO_REVIEW_TRIGGERS.items() if test(decision)]
    queue = QUEUE.get((decision.claim_type, decision.severity), 'standard_intake')
    if hits:
        return Routing(
            queue=queue,
            mode='human_review',
            reason='+'.join(hits),
            sla_hours=4 if decision.severity in ('high', 'catastrophic') else 24,
        )
    return Routing(queue=queue, mode='auto_triaged', reason='all_checks_clear', sla_hours=48)

Two properties make this safe. The triggers are fail closed: if the model is unsure, if coverage could not be confirmed, or if any fraud signal fired, a human sees it. And the routing table is data, so compliance can read it without reading Python.

Escalate on low confidence, not on vibes

Confidence is only useful if you act on it at three different layers:

1. Schema level. confidence: float = Field(ge=0, le=1) stops the model from writing “high” or 1.4.
2. Guardrail level. must_cite_sources above refuses to release a sub-threshold decision and pushes the model to say what is missing.
3. Workflow level. route() flips mode to human_review. For actions that mutate state (setting a reserve, closing a claim), add a tool guard that returns GuardrailResult.approve(): the call is held back, the run finishes with deferred tool requests, and you resume it with the adjuster’s answer. A denial reaches the model as the tool result, so it can explain itself.

Keep one rule absolute: the confidence threshold and the routing table live in your code, never in the prompt. A model that can lower its own bar is not a control.

Evaluate triage accuracy like a test suite

Pydantic Evals is code-first: a Dataset holds Cases, an experiment runs each case through your task, and Evaluators score the results. Deterministic, code-based checks are cheaper than model review, so start there:

from pydantic_evals import Case, Dataset
from pydantic_evals.evaluators import Evaluator, EvaluatorContext


class TriageInput(BaseModel):
    transcript: str
    claim_id: str | None = None


class ClaimTypeExact(Evaluator[TriageInput, TriageDecision]):
    def evaluate(self, ctx: EvaluatorContext[TriageInput, TriageDecision]) -> float:
        expected = ctx.metadata['expected_claim_type']
        return 1.0 if ctx.output.claim_type == expected else 0.0


class RouteAgreement(Evaluator[TriageInput, TriageDecision]):
    def evaluate(self, ctx: EvaluatorContext[TriageInput, TriageDecision]) -> float:
        expected_route = ctx.metadata['expected_route']
        return 1.0 if route(ctx.output).queue == expected_route else 0.0


dataset = Dataset(
    name='fnol_triage_v3',
    cases=[
        Case(
            name='rear_end_low_confidence',
            inputs=TriageInput(transcript='Rear-ended at a stop light, other driver cited.'),
            expected_output=None,
            metadata={'expected_claim_type': 'auto_collision', 'expected_route': 'fast_track'},
            evaluators=[ClaimTypeExact(), RouteAgreement()],
        ),
        Case(
            name='water_damage_exclusion',
            inputs=TriageInput(transcript='Basement flooded after storm, sewer backup endorsement unclear.'),
            metadata={'expected_claim_type': 'property_damage', 'expected_route': 'standard_intake'},
            evaluators=[ClaimTypeExact(), RouteAgreement()],
        ),
    ],
)

report = dataset.evaluate_sync(triage_task)   # your wrapped intake function
report.print(include_input=True, include_output=True, include_durations=False)

Add three layers of rigor as you grow: an LLM-as-a-judge evaluator for rationale quality (faithfulness, tone) where a fuzzy opinion is appropriate; a case with a deliberately absent clause to prove the verifier routes instead of guessing; and online evaluation, which attaches evaluators to production or staging traffic so every call or a sampled subset is graded in the background. Accuracy that only exists in your CI dataset is a number you cannot defend when the claims department asks what changed last quarter.

The audit trail

Every triage decision should be reconstructable a year later: what came in, what the model said, what documents it read, which rules fired, and who overrode it. Pydantic AI is OpenTelemetry-native, so two lines give you spans for every model and tool call:

import logfire

logfire.configure()
logfire.instrument_pydantic_ai()

Persist the structured verdict alongside it, with an immutable schema version:

{
  "decision_id": "tri_7c21",
  "claim_id": "CLM-4471902",
  "schema_version": "3",
  "corpus_version": "policies-2026-10-01",
  "model": "gpt-6-sol",
  "output": {"claim_type": "property_damage", "severity": "medium", "confidence": 0.61},
  "routing": {"queue": "property_desk", "mode": "human_review", "reason": "low_confidence"},
  "retries": 1,
  "human_override": null,
  "at": "2026-10-16T00:12:40Z"
}

Log retries and human_override deliberately. A run that needed three validation retries tells you your prompt or schema drifted, and the override field is the ground truth you use to relabel your eval dataset.

Key Takeaways

• Make the Pydantic output schema the contract: typed fields, min_length on evidence, and an explicit slot for facts the documents do not settle.
• Ground all policy facts in retrieval, then verify quotes verbatim against returned text so invented clauses become ModelRetry errors, not headlines.
• Keep routing tables and confidence thresholds in code; let the model classify but never let it choose its own bar.
• Fail closed: any fraud signal, low confidence, or unmatched exclusion routes to human review, and irreversible actions require an approval gate.
• Start evals with deterministic exact-match and route-agreement evaluators, add an LLM judge for prose quality, and sample live traffic for drift.
• Record schema version, corpus version, retries, and human overrides for every decision, and trace runs with OpenTelemetry instrumentation.

References

• https://ai.pydantic.dev
• https://pydantic.dev/docs/ai/core-concepts/output/
• https://pydantic.dev/docs/ai/harness/guardrails/
• https://pydantic.dev/docs/ai/evals/evals/