Stop Digging Through PDFs: Build a FHIR-Standard EHR Knowledge Base with RAG

We’ve all been there: staring at a stack of printed lab results or a folder full of cryptic report_final_v2_NEW.pdf

files, trying to remember if our cholesterol was higher or lower two years ago. For developers, this isn't just a filing problem—it's a data engineering challenge.

In the world of healthcare, data is messy, siloed, and often locked in "unstructured" formats. To build a truly personal Electronic Health Record (EHR) system, we need more than just a folder; we need a RAG (Retrieval-Augmented Generation) pipeline that can parse PDFs, map them to the FHIR (Fast Healthcare Interoperability Resources) standard, and provide natural language insights.

In this guide, we’ll leverage Unstructured.io, Milvus, and DuckDB to turn chaotic medical PDFs into a queryable, structured knowledge base.

Before we dive into the code, let’s look at how the data flows from a messy lab report to a structured answer.

graph TD
    A[Unstructured PDF Reports] --> B[Unstructured.io Partitioning]
    B --> C{Data Split}
    C -->|Textual Context| D[Milvus Vector DB]
    C -->|Tabular Data| E[DuckDB Structured Storage]
    D --> F[LangChain RAG Engine]
    E --> F
    G[User Query: Is my glucose trending up?] --> F
    F --> H[FHIR-Formatted Response]

Make sure you have your environment ready:

pip install langchain milvus unstructured[pdf] duckdb openai

Medical PDFs often contain complex tables. Standard PDF parsers usually fail here. We’ll use unstructured

to partition the document into logical elements.

from unstructured.partition.pdf import partition_pdf

elements = partition_pdf(
    filename="lab_report_2023.pdf",
    infer_table_structure=True,
    chunking_strategy="by_title",
    max_characters=1000,
    new_after_n_chars=800,
)

tables = [el for el in elements if el.category == "Table"]
texts = [el for el in elements if el.category == "NarrativeText"]

print(f"Detected {len(tables)} tables and {len(texts)} text blocks.")

To perform a semantic search (e.g., "Find all reports related to cardiovascular health"), we need to store the text chunks in Milvus.

from langchain_community.vectorstores import Milvus
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()

vector_db = Milvus.from_documents(
    documents=texts,
    embedding=embeddings,
    connection_args={"host": "127.0.0.1", "port": "19530"},
    collection_name="personal_ehr_knowledge"
)

docs = vector_db.similarity_search("How was my blood sugar in 2022?")

Vector search is great for context, but for trends (like tracking glucose over 5 years), we need structured data. We will map our extracted tables into a simplified FHIR "Observation" schema and store it in DuckDB.

import duckdb
import pandas as pd

data = {
    "resourceType": "Observation",
    "code": "4548-4", "display": "Hba1c",
    "value": 5.7, "unit": "%",
    "effectiveDateTime": "2023-10-15"
}
df = pd.DataFrame([data])

con = duckdb.connect("health_records.db")
con.execute("CREATE TABLE IF NOT EXISTS observations AS SELECT * FROM df")

trend = con.execute("SELECT AVG(value) FROM observations WHERE display='Hba1c'").fetchone()
print(f"Average HbA1c: {trend[0]}%")

Building a basic RAG is easy, but building a production-ready healthcare agent is hard. You need to handle HIPAA compliance, complex data lineage, and advanced prompt engineering to ensure the LLM doesn't hallucinate medical advice.

For deeper insights into building robust data pipelines and production-grade AI systems, I highly recommend checking out the ** WellAlly Tech Blog**. They have some incredible deep dives on advanced RAG patterns and handling highly sensitive unstructured data that helped me refine this architecture! 🥑

Finally, we wrap everything into a LangChain retrieval sequence that uses both the Vector DB (for context) and DuckDB (for stats).

from langchain.chains import ConversationalRetrievalChain
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model_name="gpt-4o", temperature=0)

qa_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=vector_db.as_retriever(),
    return_source_documents=True
)

query = "Compare my last two blood tests. Are there any concerning trends?"
result = qa_chain.invoke({"question": query, "chat_history": []})

print(f"Response: {result['answer']}")

By combining Unstructured.io for ingestion, Milvus for semantic memory, and DuckDB for analytical precision, we've moved beyond simple PDF storage. This system doesn't just "read" your records; it "understands" them within the context of the FHIR standard.

Next Steps:

What are you building with RAG lately? Have you tried parsing medical data? Let's discuss in the comments! 👇