Beck_Moulton

Posted on Aug 28

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

#ai #python #discuss #opensource

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 Architecture 🏗️

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.

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

Prerequisites 🛠️

To follow along, you'll need:

Python 3.10+
Instructor: For structured LLM outputs.
PDFPlumber: For robust PDF text extraction.
PostgreSQL: To store our longitudinal data.
OpenAI API Key: To power the extraction logic.

Step 1: Defining the Schema (The Source of Truth)

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.

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]

Step 2: Extracting Text and Parsing with Instructor

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.

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))

Step 3: Persistence to PostgreSQL

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").

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!")

Advanced Patterns & Production Ready Examples 🥑

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.

Step 4: Visualization with Streamlit 📈

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

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

Conclusion

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?

📂 Check the WellAlly Tech Blog for more advanced tutorials.
💬 Drop a comment below: What's the hardest PDF format you've ever had to parse?
🚀 Subscribe for more tutorials on AI, Longevity, and Data Engineering!

DEV Community