PydanticAI

Overview

Why monitor PydanticAI?

PydanticAI's structured output extraction is powerful — but it can fail in ways that are hard to trace. The model might return a field in the wrong type. A required field might be missing. Retries might exhaust silently. In high-volume pipelines (hundreds of documents per hour), even a 2% validation failure rate means dozens of silently dropped records.

NotiLens tracks every extraction attempt — model call start, structured output validated, validation failed, and token cost — so you know immediately when your extraction pipeline degrades.

Auto-Patch

Installation

pip install notilens pydantic-ai

Setup

Add patch=True — NotiLens hooks into PydanticAI's execution and tracks every model call and structured output event.

import notilens
from pydantic_ai import Agent
from pydantic import BaseModel

nl = notilens.init(
    name="extraction-pipeline",
    token="YOUR_TOKEN",
    secret="YOUR_SECRET",
    patch=True
)

class FeedbackCategory(BaseModel):
    sentiment:  str          # positive / negative / neutral
    topic:      str          # billing / product / support / other
    priority:   str          # high / medium / low
    summary:    str
    action_required: bool

agent = Agent(
    "openai:gpt-4o-mini",
    result_type=FeedbackCategory,
    system_prompt="""Categorise customer feedback.
    sentiment: positive/negative/neutral
    priority: high if they mention churn or legal""",
)

# NotiLens tracks: model call, structured output validation, tokens + cost
result = agent.run_sync(
    "Your product deleted all my data and support took 3 days to respond."
)
print(result.data.sentiment)   # negative
print(result.data.priority)    # high

What gets tracked automatically

PydanticAI's validation step is captured as a separate event — so you can distinguish between a model failure (the LLM errored) and a validation failure (the LLM succeeded but returned the wrong shape).

Real-world example — invoice data extraction

Extracting structured data from raw invoice text. NotiLens alerts when the model fails to extract a required field (like total_amount) — a common failure when invoice formats vary. Over time, it baselines your extraction accuracy and alerts when the failure rate spikes.

import notilens
from pydantic_ai import Agent
from pydantic import BaseModel
from decimal import Decimal

nl = notilens.init(
    name="invoice-extractor",
    token="YOUR_TOKEN",
    secret="YOUR_SECRET",
    patch=True
)

class InvoiceData(BaseModel):
    vendor_name:    str
    invoice_number: str
    invoice_date:   str          # ISO format YYYY-MM-DD
    total_amount:   Decimal
    currency:       str          # ISO 4217 e.g. USD, EUR
    line_items:     list[dict]   # [{"description": ..., "amount": ...}]
    payment_due:    str | None   # ISO date or None

agent = Agent(
    "openai:gpt-4o-mini",
    result_type=InvoiceData,
    system_prompt="""Extract invoice data exactly as specified.
    For dates use YYYY-MM-DD format. For amounts use decimal numbers only.""",
)

# Process invoices in batch — NotiLens tracks every extraction
invoices = load_invoices_from_s3(bucket="incoming-invoices")

for invoice in invoices:
    # Each call tracked: model call → validation → output.generated or output.failed
    result = agent.run_sync(invoice.raw_text)
    save_to_accounting_system(result.data)

# NotiLens will alert if:
# - output.failed rate suddenly increases (new invoice format breaking extraction)
# - Token cost spikes (unusually long invoice text)
# - Latency increases (model degradation)

Manual

Manual Instrumentation

Wrap PydanticAI extraction in a task run when you need to track the full pipeline — document fetching, extraction, validation, storage — not just the model call.

Real-world use: a medical records extraction pipeline where a failed extraction needs to be flagged for manual review by a clinician, not silently dropped.

import notilens
from pydantic_ai import Agent
from pydantic import BaseModel

nl  = notilens.init(name="medical-extractor", token="YOUR_TOKEN", secret="YOUR_SECRET")
run = nl.task("record-extraction")
run.start()

try:
    run.progress(f"Fetching record {record_id}")
    raw_text = fetch_record(record_id)

    run.progress("Extracting structured data")
    result = agent.run_sync(raw_text)

    # Track extraction quality metrics
    run.metric("tokens",          result.usage().total_tokens)
    run.metric("fields_extracted", len(result.data.model_fields_set))

    # Flag incomplete extractions for manual review
    missing = [f for f in required_fields if f not in result.data.model_fields_set]
    if missing:
        run.output_generated(f"Partial extraction — {len(missing)} fields missing")
        run.input_required(
            f"Clinician review needed: {record_id} — missing {', '.join(missing)}"
        )
        decision = wait_for_clinical_review(record_id)
        if decision["approved"]:
            run.input_approved("Clinician filled missing fields")
            save_with_manual_fields(result.data, decision["additions"])
            run.complete(f"Record {record_id} processed")
        else:
            run.input_rejected("Record flagged — cannot process")
    else:
        run.output_generated(f"Full extraction — all {len(required_fields)} fields present")
        save_record(result.data)
        run.complete(f"Record {record_id} saved")

except Exception as e:
    run.fail(str(e))

Monitoring Validation Failures

Detecting extraction degradation

In high-volume extraction pipelines, the most useful signal is when your validation failure rate changes — not just when individual extractions fail. NotiLens baselines your failure rate and alerts on anomalies.

Example: you process 500 invoices per day with a 1.5% validation failure rate. A new vendor starts sending PDFs in an unusual format — failure rate jumps to 12%. NotiLens detects the spike before you've manually noticed 60 failed extractions.

import notilens
from pydantic_ai import Agent, ModelRetry

nl  = notilens.init(name="bulk-extractor", token="YOUR_TOKEN", secret="YOUR_SECRET")

success_count = 0
failure_count = 0

for doc in documents:
    run = nl.task("extract")
    run.start()
    try:
        result = agent.run_sync(doc.text)
        success_count += 1
        run.metric("tokens", result.usage().total_tokens)
        run.output_generated("Extraction successful")
        run.complete("OK")
    except Exception as e:
        failure_count += 1
        run.output_failed(f"Extraction failed: {type(e).__name__}")
        run.fail(str(e))

# End-of-batch summary — NotiLens learns your normal failure rate
nl.notify("batch.complete",
    f"Processed {len(documents)} docs — {success_count} success, {failure_count} failed",
    meta={
        "success_count":  success_count,
        "failure_count":  failure_count,
        "failure_rate":   round(failure_count / len(documents) * 100, 2),
    }
)