# ETL for Longevity: Automating Blood Test Analysis with GPT-4 and Instructor

> Source: <https://dev.to/beck_moulton/etl-for-longevity-automating-blood-test-analysis-with-gpt-4-and-instructor-37eh>
> Published: 2026-08-28 00:36:00+00:00

We’ve all been there: you get your annual blood work done, and the hospital sends you a cryptic 5-page PDF. You want to track your glucose or cholesterol levels over time, but the data is trapped in a non-standardized table format. Doing this manually is a nightmare.

In the world of **Data Engineering**, this is a classic "unstructured to structured" problem. Today, we are building a robust **LLM-powered ETL pipeline** to automate the ingestion of blood test reports into a **PostgreSQL** database. By leveraging **Structured Outputs** with the `Instructor`

library, we can turn messy PDFs into clean, queryable health insights.

Whether you are building a personal longevity dashboard or a professional health informatics tool, mastering this **structured data extraction** workflow is a game-changer for your **health data pipeline**.

The flow is straightforward but powerful. We extract the raw text, pass it to a Large Language Model (LLM) constrained by a strict Pydantic schema, and then commit that validated data to our database.

``` php
graph TD
    A[Blood Test PDF] -->|PDFPlumber| B(Raw Text Extraction)
    B -->|Text Prompt| C{Instructor + GPT-4o}
    C -->|Validated JSON| D[Pydantic Model]
    D -->|SQL Insert| E[(PostgreSQL Database)]
    E -->|Query| F[Streamlit Dashboard]

    subgraph "Validation Layer"
    C
    D
    end
```

To follow along, you'll need:

The secret sauce to a reliable ETL is a strict schema. We use `Pydantic`

to define exactly what a "Blood Test" looks like. This ensures the LLM doesn't hallucinate random units or field names.

``` python
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import date

class BloodMarker(BaseModel):
    name: str = Field(..., description="Common name of the marker, e.g., LDL Cholesterol")
    value: float = Field(..., description="The numerical value recorded")
    unit: str = Field(..., description="The unit of measurement, e.g., mg/dL")
    reference_range: Optional[str] = Field(None, description="The normal range provided by the lab")

class BloodReport(BaseModel):
    report_date: date
    hospital_name: str
    patient_name: str
    markers: List[BloodMarker]
```

Hospital PDFs are notoriously inconsistent. Some use tables, others use key-value columns. We use `PDFPlumber`

to get the raw text and let `Instructor`

handle the heavy lifting of understanding the context.

``` python
import pdfplumber
import instructor
from openai import OpenAI

# Initialize Instructor-patched client
client = instructor.from_openai(OpenAI())

def extract_blood_data(pdf_path: str) -> BloodReport:
    # 1. Extract raw text from PDF
    with pdfplumber.open(pdf_path) as pdf:
        raw_text = "\n".join([page.extract_text() for page in pdf.pages])

    # 2. Use Instructor to parse the text into our Pydantic model
    report = client.chat.completions.create(
        model="gpt-4o",
        response_model=BloodReport,
        messages=[
            {"role": "system", "content": "You are a specialized medical data extractor."},
            {"role": "user", "content": f"Extract all blood markers from this text: {raw_text}"}
        ],
    )
    return report

# Usage
# data = extract_blood_data("my_blood_work_2023.pdf")
# print(data.model_dump_json(indent=2))
```

Once we have a validated `BloodReport`

object, saving it to a relational database allows us to perform time-series analysis (e.g., "Show me my Vitamin D levels over the last 3 years").

``` python
import psycopg2
from psycopg2.extras import execute_values

def save_to_db(report: BloodReport):
    conn = psycopg2.connect("dbname=longevity_db user=postgres password=secret")
    cur = conn.cursor()

    # Simple logic to insert markers
    query = """
    INSERT INTO blood_results (report_date, marker_name, value, unit)
    VALUES %s
    """
    data_points = [
        (report.report_date, m.name, m.value, m.unit) 
        for m in report.markers
    ]

    execute_values(cur, query, data_points)
    conn.commit()
    cur.close()
    conn.close()
    print("🚀 Data successfully ingested!")
```

While this script works for individual files, production-grade longevity apps require handling multi-page tables, OCR for scanned images, and data normalization (e.g., converting 'mg/dL' to 'mmol/L').

For a deeper dive into production-ready AI architectures and advanced data engineering patterns for health tech, I highly recommend checking out the ** WellAlly Tech Blog**. They cover extensively how to scale these LLM workflows and handle complex medical data privacy requirements.

Finally, let's wrap this in a simple UI. Streamlit allows us to upload a PDF and immediately see our historical trends.

``` python
import streamlit as st

st.title("Longevity Tracker 🧬")
uploaded_file = st.file_uploader("Upload Blood Test PDF", type="pdf")

if uploaded_file:
    with st.spinner("Processing medical data..."):
        # Logic to extract and save...
        st.success("Analysis Complete!")
        # Use st.line_chart to show historical data from SQL
```

By combining **Instructor** for structured LLM outputs and **PostgreSQL** for long-term storage, we've turned a manual, error-prone task into a seamless automated pipeline. This is the foundation of "Learning in Public"—taking a personal pain point (messy medical PDFs) and solving it with a modern tech stack.

**Ready to take your data engineering to the next level?**