Extracting structured fields from free text with an LLM

# ======================================================================
# Q: We want an LLM to read free text (e.g. vendor advisories) and produce
# structured fields. How do you build it?
#
# The LLM is just another external, slow and non-deterministic API, so
# everything I know about integrations applies, plus validation of the result. I
# hide the provider behind our own interface so the model or provider can be
# swapped.
# Signs of maturity: structured output with a schema and validation, async in
# Celery with rate limit and retries, a cache keyed by the hash of the input
# (cost), recording the model and prompt version on every result
# (audit/compliance), human review for anything that fails or has low
# confidence, and measuring tokens/cost per tenant.
# ======================================================================

class AdvisoryExtract(BaseModel):
    cve_ids: list[str] = Field(default_factory=list)
    affected_products: list[str]
    severity: Literal['low', 'medium', 'high', 'critical']
    summary: str = Field(max_length=500)

    @field_validator('cve_ids')
    @classmethod
    def valid_cves(cls, v):
        return [c for c in v if re.fullmatch(r'CVE-\d{4}-\d{4,}', c)]

@shared_task(bind=True, autoretry_for=(LLMTimeout, LLMRateLimited),
             retry_backoff=True, max_retries=5, rate_limit='30/m')
def extract_advisory(self, doc_id):
    doc = load_document(doc_id)
    cache_key = f'llm:advisory:v3:{sha256(doc.text)}'   # same text -> same result
    if cached := cache.get(cache_key):
        return save_extract(doc_id, AdvisoryExtract.model_validate(cached))
    raw = llm.complete_json(
        system=EXTRACT_PROMPT_V3,                       # versioned prompt
        user=wrap_untrusted(doc.text[:50_000]),         # data, not instructions
        schema=AdvisoryExtract.model_json_schema(),
        timeout=60)
    try:
        result = AdvisoryExtract.model_validate(raw)
    except ValidationError:
        mark_for_human_review(doc_id, raw)               # is not accepted silently
        return
    cache.set(cache_key, result.model_dump(), timeout=30 * 86400)
    METRIC_LLM_TOKENS.labels(task='advisory').inc(raw.usage_total)
    save_extract(doc_id, result, model=llm.model_name, prompt_version=3)