Q&A over patient intake forms with LangChain

Patient intake forms contain dense, unstructured clinical data that LLMs can unlock — if you handle document ingestion, chunking, and retrieval correctly. This tutorial walks through building a complete Q&A pipeline over intake forms using LangChain, from raw PDFs to a queryable chain with source attribution. We’ll cover PHI considerations, chunking strategies for medical forms, and a runnable implementation you can extend.

Prerequisites

• Python 3.10+
• An OpenAI API key (or compatible endpoint) for embeddings and chat
• LangChain 0.1+ and supporting packages
• Sample patient intake forms as PDFs or text files

Install dependencies:

pip install langchain langchain-openai langchain-community pypdf faiss-cpu python-dotenv tiktoken

Create a .env file:

OPENAI_API_KEY=sk-...

Project structure

intake-qa/
├── data/
│   └── intake_forms/          # Drop PDFs here
├── src/
│   ├── ingest.py              # Document loading and indexing
│   ├── chain.py               # QA chain construction
│   └── query.py               # CLI entry point
├── .env
└── requirements.txt

Loading and chunking intake forms

Medical forms have distinct sections — demographics, history of present illness, review of systems, medications, allergies. Naive chunking splits mid-section and loses context. We’ll use a recursive splitter with healthcare-aware separators.

# src/ingest.py
import os
from pathlib import Path
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from dotenv import load_dotenv

load_dotenv()

DATA_DIR = Path(__file__).parent.parent / "data" / "intake_forms"
INDEX_DIR = Path(__file__).parent.parent / "index" / "faiss_intake"

SEPARATORS = [
    "\n\n## ",           # Markdown-style section headers
    "\n\n",              # Paragraph breaks
    "\n",                # Line breaks
    ". ",                # Sentences
    " ",                 # Words
    "",                  # Characters
]

def load_documents():
    docs = []
    for pdf_path in DATA_DIR.glob("*.pdf"):
        loader = PyPDFLoader(str(pdf_path))
        loaded = loader.load()
        for d in loaded:
            d.metadata["source_file"] = pdf_path.name
        docs.extend(loaded)
    return docs

def chunk_documents(docs):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=800,
        chunk_overlap=120,
        separators=SEPARATORS,
        length_function=len,
    )
    return splitter.split_documents(docs)

def build_index(chunks):
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vs = FAISS.from_documents(chunks, embeddings)
    INDEX_DIR.parent.mkdir(parents=True, exist_ok=True)
    vs.save_local(str(INDEX_DIR))
    return vs

if __name__ == "__main__":
    print("Loading documents...")
    docs = load_documents()
    print(f"Loaded {len(docs)} pages from {len(set(d.metadata['source_file'] for d in docs))} files")

    print("Chunking...")
    chunks = chunk_documents(docs)
    print(f"Created {len(chunks)} chunks")

    print("Building FAISS index...")
    build_index(chunks)
    print(f"Index saved to {INDEX_DIR}")

Run it:

python -m src.ingest

Expected output:

Loading documents...
Loaded 12 pages from 3 files
Chunking...
Created 47 chunks
Building FAISS index...
Index saved to /path/to/index/faiss_intake

Building the QA chain

We’ll use a retrieval-augmented generation (RAG) chain with source citations. The prompt instructs the model to only answer from retrieved context and to flag when information is missing.

# src/chain.py
from pathlib import Path
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv

load_dotenv()

INDEX_DIR = Path(__file__).parent.parent / "index" / "faiss_intake"

SYSTEM_PROMPT = """You are a clinical assistant answering questions about patient intake forms.
Use only the provided context to answer. If the context does not contain the answer, say you don't know.
Cite sources using the format [source: filename, page N] at the end of each sentence.
Be concise and clinical. Do not hallucinate medications, allergies, or diagnoses."""

PROMPT = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_PROMPT),
    ("human", "Context:\n{context}\n\nQuestion: {question}"),
])

def format_docs(docs):
    lines = []
    for d in docs:
        src = d.metadata.get("source_file", "unknown")
        page = d.metadata.get("page", "?")
        lines.append(f"[source: {src}, page {page}]\n{d.page_content}")
    return "\n\n---\n\n".join(lines)

def get_chain(k=6):
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vs = FAISS.load_local(str(INDEX_DIR), embeddings, allow_dangerous_deserialization=True)
    retriever = vs.as_retriever(search_kwargs={"k": k})

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

    rag_chain = (
        RunnableParallel({
            "context": retriever | format_docs,
            "question": RunnablePassthrough(),
        })
        | PROMPT
        | llm
        | StrOutputParser()
    )
    return rag_chain

Query CLI with source display

# src/query.py
import sys
from src.chain import get_chain

def main():
    if len(sys.argv) < 2:
        print("Usage: python -m src.query \"your question here\"")
        sys.exit(1)

    question = " ".join(sys.argv[1:])
    chain = get_chain()

    print(f"Question: {question}\n")
    print("Answer:")
    for chunk in chain.stream(question):
        print(chunk, end="", flush=True)
    print()

if __name__ == "__main__":
    main()

Test it:

python -m src.query "What medications is the patient currently taking?"

Expected output:

Question: What medications is the patient currently taking?

Answer: The patient is currently taking lisinopril 10 mg daily, metformin 500 mg twice daily, and atorvastatin 20 mg at bedtime [source: intake_john_doe.pdf, page 2].

Handling PHI and PII

Patient intake forms contain protected health information. Before sending chunks to any external LLM, you should de-identify or use a local model. Two practical approaches:

Option 1: Local embeddings + local LLM

Swap OpenAIEmbeddings for a local model and run inference on-prem:

# In ingest.py and chain.py
from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2",
    model_kwargs={"device": "cpu"},
)

Pair with a local LLM via Ollama or vLLM:

from langchain_community.llms import Ollama

llm = Ollama(model="llama3", temperature=0)

Option 2: Redact before indexing

Use a NER-based redaction pass during ingestion. presidio-analyzer and presidio-anonymizer work well for this:

pip install presidio-analyzer presidio-anonymizer

# Add to ingest.py before chunking
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

def redact_text(text: str) -> str:
    results = analyzer.analyze(text=text, entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "DATE_TIME", "LOCATION", "MEDICAL_LICENSE"], language="en")
    return anonymizer.anonymize(text=text, analyzer_results=results).text

# In load_documents(), after loading each page:
for d in loaded:
    d.page_content = redact_text(d.page_content)

This replaces names, dates, and identifiers with tokens like <PERSON>, <DATE_TIME> before embeddings are created. The tradeoff: you lose the ability to answer “What is the patient’s name?” but gain safety for external API calls.

Improving retrieval for clinical forms

Default similarity search works, but clinical queries benefit from two enhancements:

Hybrid search (keyword + vector)

FAISS supports hybrid via FAISS.as_retriever with a custom search function, or use a dedicated hybrid store like Weaviate or Pinecone. A lightweight alternative: combine BM25 with vector scores.

# In chain.py
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from typing import List

class HybridRetriever(BaseRetriever):
    vector_retriever: BaseRetriever
    bm25_retriever: BM25Retriever
    k: int = 6

    def _get_relevant_documents(self, query: str) -> List[Document]:
        vec_docs = self.vector_retriever.invoke(query)
        bm25_docs = self.bm25_retriever.invoke(query)
        # Simple reciprocal rank fusion
        seen = set()
        fused = []
        for i, d in enumerate(vec_docs + bm25_docs):
            key = d.metadata.get("source_file", "") + str(d.metadata.get("page", "")) + d.page_content[:50]
            if key not in seen:
                seen.add(key)
                fused.append(d)
            if len(fused) >= self.k:
                break
        return fused

Initialize it in get_chain():

def get_chain(k=6):
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vs = FAISS.load_local(str(INDEX_DIR), embeddings, allow_dangerous_deserialization=True)
    vector_retriever = vs.as_retriever(search_kwargs={"k": k * 2})

    # Build BM25 from the same chunks
    all_docs = list(vs.docstore._dict.values())
    bm25 = BM25Retriever.from_documents(all_docs)
    bm25.k = k * 2

    retriever = HybridRetriever(vector_retriever=vector_retriever, bm25_retriever=bm25, k=k)
    # ... rest unchanged

Query expansion for clinical synonyms

Patients and clinicians use different terms: “high blood pressure” vs “hypertension”, “heart attack” vs “myocardial infarction”. Expand the query before retrieval:

from langchain_core.runnables import RunnableLambda

CLINICAL_SYNONYMS = {
    "high blood pressure": "hypertension",
    "heart attack": "myocardial infarction",
    "stroke": "cerebrovascular accident",
    "blood sugar": "glucose",
    "kidney failure": "renal failure",
}

def expand_query(query: str) -> str:
    expanded = query
    for lay, clinical in CLINICAL_SYNONYMS.items():
        if lay.lower() in query.lower():
            expanded += f" {clinical}"
    return expanded

# In get_chain():
retriever = (
    RunnableLambda(lambda q: expand_query(q))
    | retriever
)

Evaluation checkpoint

Before deploying, run a small eval set. Create eval/questions.jsonl:

{"question": "What is the patient's primary care physician?", "expected_source": "intake_john_doe.pdf"}
{"question": "List all documented allergies.", "expected_source": "intake_john_doe.pdf"}
{"question": "What was the date of the last colonoscopy?", "expected_source": "intake_jane_smith.pdf"}

Evaluate with a simple script:

# src/eval.py
import json
from src.chain import get_chain

def evaluate():
    chain = get_chain()
    with open("eval/questions.jsonl") as f:
        for line in f:
            item = json.loads(line)
            answer = chain.invoke(item["question"])
            print(f"Q: {item['question']}")
            print(f"A: {answer[:200]}...")
            print(f"Expected source: {item['expected_source']}")
            print("---")

if __name__ == "__main__":
    evaluate()

Look for: correct answers grounded in cited sources, appropriate “I don’t know” when info is missing, and consistent citation format.

Production considerations

Incremental updates

New intake forms arrive daily. Rebuild the index nightly or use FAISS’s add_documents:

def add_new_forms(vs, new_chunks):
    vs.add_documents(new_chunks)
    vs.save_local(str(INDEX_DIR))

Multi-tenant isolation

If serving multiple clinics, namespace the vector store per tenant:

# Index path includes tenant_id
INDEX_DIR = Path(__file__).parent.parent / "index" / f"faiss_intake_{tenant_id}"

Pass tenant_id through the chain and load the correct index.

Routing and fallback

When using multiple model providers, route clinical queries to a model with strong medical knowledge and fall back if degraded. If you’re using a gateway like n4n.ai, you can specify routing directives in the request headers to prefer certain models for clinical workloads while maintaining a single OpenAI-compatible endpoint.

Monitoring

Log every query with: question hash, retrieved chunk IDs, latency, token usage, and a user feedback signal (thumbs up/down). This lets you measure retrieval quality and catch drift.

Next steps

• Add a FastAPI wrapper for HTTP access
• Implement role-based access (clinician vs admin views)
• Integrate with FHIR resources for structured data linkage
• Add a human-in-the-loop review queue for low-confidence answers
• Experiment with re-ranking (Cohere Rerank, bge-reranker) for tighter context

The pipeline above is deliberately minimal — no framework magic, just composable LangChain primitives you can inspect and replace. Start here, measure, then add complexity where your evals show it matters.