Respiratory Disease NLP + Bayesian Network
Build a clinical AI pipeline that diagnoses respiratory diseases from physician notes.
Introduction
30 Second Summary
When a physician writes notes during a patient visit, those notes contain critical diagnostic clues buried in unstructured text. Hospitals need systems that can automatically extract these clues and reason about what disease a patient might have.
In this project, you will build a Python pipeline that reads synthetic clinical notes, extracts respiratory findings using NLP, and feeds the structured evidence into a Bayesian network that computes disease probabilities.
What You'll Build
A clinical AI system that chains NLP processing stages with probabilistic inference to transform raw physician notes into ranked respiratory disease diagnoses.
By the end of this project, you'll have:
- A pipeline orchestrator that chains section detection, finding extraction, and negation handling into a single processing flow.
- A Bayesian network built with pgmpy that computes disease probabilities from extracted clinical evidence.
- A test suite that validates each pipeline stage independently and as an integrated whole.
- Secret Mission: A confidence calibration feature that evaluates your model's predictions against known outcomes and reports where it is overconfident or underconfident.
Are there any prerequisites?
This project assumes you are already familiar with the individual concepts it integrates: regex, dataclasses, unit testing, conditional probability, and pgmpy. The focus is on connecting these pieces into a cohesive system, not on first exposure to any single concept.
You will also need basic Python proficiency and a working Python 3.10+ environment.
Before We Start
This project builds a multi-stage clinical NLP pipeline that feeds extracted evidence into a Bayesian network for respiratory disease diagnosis.
Before writing any pipeline code, you need Python, two key libraries, and a project folder with a module file for each pipeline stage.
In this step, get ready to:
- Install and verify Python 3.10+.
- Install pgmpy and pytest via pip.
- Create the project folder structure with separate modules for each pipeline stage.
Install and verify Python 3.10+
The pgmpy library requires Python 3.10 or higher. Let's check whether you already have a compatible version installed.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
- Check your Python version by running this command:
macOS
python3 --versionWindows
python --version✔️ I see version 3.10 or higher
You're all set. Move on to the next substep.
ⓧ I see an older version
Your system has a Python version below 3.10. macOS often ships with an older Python by default, so even after installing a newer version, the python3 command may still point to the old one.
- Download the latest Python from python.org/downloads.
- Run the installer and follow the prompts.
- After installation, close and reopen your terminal.
- Run the version check again to confirm you see 3.10 or higher.
macOS still showing the old version?
Try running python3.12 --version (or whichever version you installed) using the full version-specific binary name. If that works, use python3.12 and pip3.12 throughout this project instead of python3 and pip.
ⓧ Command not found
Python is not installed on your machine yet.
- Download Python from python.org/downloads.
- Run the installer. On Windows, make sure you check the Add python.exe to PATH checkbox before clicking Install Now.
- Close and reopen your terminal after installation.
- Run the version check command again to confirm Python 3.10+ is available.
Install pgmpy and pytest
You need two libraries: pgmpy for building Bayesian networks, and pytest for running your test suite.
- Install both libraries by running this command:
macOS
pip3 install pgmpy pytestWindows
pip install pgmpy pytestWhat are these libraries?
pgmpy is a Python library for probabilistic graphical models. You will use it to define a Bayesian network structure, attach conditional probability tables, and run inference queries.
pytest is a testing framework. You will use it to validate each pipeline stage independently before wiring everything together.
- Verify the installations succeeded by running:
macOS
python3 -c "import pgmpy; print(pgmpy.__version__)"Windows
python -c "import pgmpy; print(pgmpy.__version__)"You should see a version number printed (such as 1.1.2).
Seeing a ModuleNotFoundError?
- Make sure pip matches your Python version. If you're using python3.12, run pip3.12 install pgmpy pytest instead.
- On some systems, you may need to use pip3 instead of pip.
Help me troubleshoot my pgmpy installation error.
Create the project folder structure
Each pipeline stage lives in its own Python module. This keeps the codebase organized so you can develop and test each stage independently.
- Move to your Desktop by running this command:
macOS
cd ~/DesktopWindows
cd ~/Desktop- Create your project folder and move into it by running these commands:
macOS
mkdir respiratory_pipeline
cd respiratory_pipelineWindows
mkdir respiratory_pipeline
cd respiratory_pipeline- Create the empty module files for each pipeline stage by running this command:
macOS
touch models.py section_detector.py finding_extractor.py negation_detector.py pipeline.py cpt_builder.py bayesian_network.py synthetic_data.py test_pipeline.py main.pyWindows
New-Item -ItemType "File" -Path "models.py", "section_detector.py", "finding_extractor.py", "negation_detector.py", "pipeline.py", "cpt_builder.py", "bayesian_network.py", "synthetic_data.py", "test_pipeline.py", "main.py"- Confirm the files exist by running:
lsYou should see all 10 .py files listed: models.py, section_detector.py, finding_extractor.py, negation_detector.py, pipeline.py, cpt_builder.py, bayesian_network.py, synthetic_data.py, test_pipeline.py, and main.py.
What is each file for?
- models.py holds typed dataclasses for structured findings.
- section_detector.py splits clinical notes into labeled sections.
- finding_extractor.py extracts respiratory findings using regex.
- negation_detector.py identifies negated findings in clinical language.
- pipeline.py orchestrates all NLP stages in sequence.
- cpt_builder.py computes conditional probability tables from extracted data.
- bayesian_network.py defines and runs inference on the pgmpy network.
- synthetic_data.py stores synthetic clinical notes for training and testing.
- test_pipeline.py contains your pytest test suite.
- main.py wires everything together into the diagnose() function.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type VS Code and press Enter to open it.
- In VS Code, click File in the top menu bar.
- Click Open Folder.
- Navigate to your Desktop, select the respiratory_pipeline folder, and click Open.
You should see all 10 .py files listed in the Explorer sidebar on the left.
Don't see the files in VS Code?
- Make sure you opened the respiratory_pipeline folder itself, not the Desktop folder.
- If the Explorer sidebar is collapsed, click the top icon in the left sidebar (two overlapping files) to expand it.
Help me open my project folder in VS Code.
Your environment is ready. Next up, you will build the pipeline orchestrator that chains all processing stages together.
Build the Typed Dataclass Models
Before building any processing logic, the pipeline needs a clear data contract. Every stage in the system needs to know exactly what shape data takes as it flows from one component to the next.
In this step, you'll define Python dataclasses that represent clinical findings, note sections, and complete pipeline output. These typed models act as the shared vocabulary that keeps every pipeline stage aligned.
In this step, get ready to:
- Define ClinicalFinding and SectionResult dataclasses with typed fields.
- Build a PipelineOutput dataclass with filtering and serialization methods.
- Verify the models work by instantiating them in Python.
Define the ClinicalFinding and SectionResult dataclasses
The pipeline will extract individual clinical findings (like "wheezing" or "consolidation") and detect sections of a clinical note (like "History of Present Illness"). Each of these needs a structured representation with typed fields.
- Open models.py in the Explorer sidebar of VS Code.
- Add the following code to define your first two data models:
"""Typed dataclass models for the clinical NLP pipeline."""
from dataclasses import dataclass, field
@dataclass
class ClinicalFinding:
"""A single clinical finding extracted from a note."""
# Core identification
name: str
section: str
is_negated: bool
# Where in the original text this finding appeared
text_span: str
start_pos: int
end_pos: int
def to_evidence_key(self) -> str:
# Convert finding name to a BN-compatible key (e.g. "decreased breath sounds" -> "Decreased_Breath_Sounds")
return self.name.replace(" ", "_").title()
def to_evidence_value(self) -> str:
# Map negation status to BN evidence values
return "absent" if self.is_negated else "present"
@dataclass
class SectionResult:
"""A detected section from a clinical note."""
name: str
content: str
start_pos: int
end_pos: intWhat does this code do?
ClinicalFinding represents a single respiratory symptom extracted from a clinical note. The is_negated field tracks whether the finding was denied ("no wheezing") or affirmed ("bilateral wheezing").
The two helper methods (to_evidence_key and to_evidence_value) convert findings into the format the Bayesian network expects later in the project.
SectionResult represents a labeled section of a clinical note (like HPI or Physical Exam). The position fields let downstream stages know exactly where each section lives in the original text.
Why dataclasses instead of plain dictionaries?
Dictionaries let you store anything with any key. Dataclasses enforce a specific shape. If a downstream stage expects is_negated to be a bool, the type annotation makes that contract explicit.
This prevents subtle bugs where one stage writes a string and another reads a boolean. In a multi-stage pipeline, clear contracts between stages save hours of debugging.
Build the PipelineOutput dataclass with helper methods
The pipeline needs a single container that accumulates results from every processing stage. This container also needs helper methods that filter and serialize findings for the Bayesian network.
- In models.py, add the following code below the SectionResult class:
@dataclass
class PipelineOutput:
"""Complete output from the NLP pipeline."""
raw_note: str
# Accumulate results from each stage (default to empty lists)
sections: list[SectionResult] = field(default_factory=list)
findings: list[ClinicalFinding] = field(default_factory=list)
errors: list[str] = field(default_factory=list)
def get_affirmed_findings(self) -> list[ClinicalFinding]:
# Filter to only findings that were NOT negated
return [f for f in self.findings if not f.is_negated]
def get_negated_findings(self) -> list[ClinicalFinding]:
# Filter to only findings that WERE negated
return [f for f in self.findings if f.is_negated]
def to_evidence_dict(self) -> dict[str, str]:
# Convert all findings into a dict the Bayesian network can consume
evidence = {}
for finding in self.findings:
key = finding.to_evidence_key()
value = finding.to_evidence_value()
evidence[key] = value
return evidenceWhat does this code do?
PipelineOutput is the master container for everything the pipeline produces. It stores the original note, detected sections, extracted findings, and any errors that occurred during processing.
field(default_factory=list) ensures each new instance gets its own fresh list. Without this, all instances would accidentally share the same list object.
The three helper methods let you quickly filter findings by negation status or serialize everything into the dictionary format the Bayesian network needs for inference.
- Save models.py.
Seeing a red underline on list[SectionResult]?
The list[SectionResult] syntax requires Python 3.10+. If your editor shows an error, confirm your Python version by running python3 --version in your terminal.
Also check that ClinicalFinding and SectionResult are defined above PipelineOutput in the file. Python reads top-to-bottom, so the referenced classes must exist first.
Help me fix a type annotation error in my dataclass models.
Verify the models in the Python REPL
A quick sanity check confirms that your dataclasses instantiate correctly and the helper methods produce expected output.
- Open your terminal in VS Code by pressing Ctrl+` (backtick key, top-left of your keyboard).
- Start a Python REPL and test your models by running:
python3 -c "
from models import ClinicalFinding, SectionResult, PipelineOutput
# Create a test finding
f = ClinicalFinding(name='wheezing', section='PE', is_negated=False, text_span='wheezing', start_pos=10, end_pos=18)
print(f'Finding: {f.name}, Key: {f.to_evidence_key()}, Value: {f.to_evidence_value()}')
# Create a PipelineOutput with one affirmed and one negated finding
neg = ClinicalFinding(name='fever', section='HPI', is_negated=True, text_span='fever', start_pos=5, end_pos=10)
output = PipelineOutput(raw_note='test', findings=[f, neg])
print(f'Affirmed: {len(output.get_affirmed_findings())}, Negated: {len(output.get_negated_findings())}')
print(f'Evidence dict: {output.to_evidence_dict()}')
"- You should see output like this:
Finding: wheezing, Key: Wheezing, Value: present
Affirmed: 1, Negated: 1
Evidence dict: {'Wheezing': 'present', 'Fever': 'absent'}What should I see?
The output confirms three things. First, to_evidence_key() correctly converts "wheezing" to "Wheezing" (title-cased, underscores for spaces).
Second, get_affirmed_findings() and get_negated_findings() correctly partition findings by their negation status. Third, to_evidence_dict() produces the exact format the Bayesian network will consume later.
ModuleNotFoundError: No module named 'models'?
Make sure your terminal is in the project folder that contains models.py. Run ls to confirm you can see the file listed in the current directory.
Help me fix this ModuleNotFoundError when importing my models.
✔️ Awesome, I've got everything!
Great. Double check you've saved models.py.
ⓧ I'd like to double check the full code
"""Typed dataclass models for the clinical NLP pipeline."""
from dataclasses import dataclass, field
@dataclass
class ClinicalFinding:
"""A single clinical finding extracted from a note."""
name: str
section: str
is_negated: bool
text_span: str
start_pos: int
end_pos: int
def to_evidence_key(self) -> str:
return self.name.replace(" ", "_").title()
def to_evidence_value(self) -> str:
return "absent" if self.is_negated else "present"
@dataclass
class SectionResult:
"""A detected section from a clinical note."""
name: str
content: str
start_pos: int
end_pos: int
@dataclass
class PipelineOutput:
"""Complete output from the NLP pipeline."""
raw_note: str
sections: list[SectionResult] = field(default_factory=list)
findings: list[ClinicalFinding] = field(default_factory=list)
errors: list[str] = field(default_factory=list)
def get_affirmed_findings(self) -> list[ClinicalFinding]:
return [f for f in self.findings if not f.is_negated]
def get_negated_findings(self) -> list[ClinicalFinding]:
return [f for f in self.findings if f.is_negated]
def to_evidence_dict(self) -> dict[str, str]:
evidence = {}
for finding in self.findings:
key = finding.to_evidence_key()
value = finding.to_evidence_value()
evidence[key] = value
return evidenceYour data models are locked in. Every pipeline stage now has a clear contract for the data it produces and consumes. Next up, you'll build the section detector that splits raw clinical notes into labeled SectionResult objects.
Build the Section Detection System
Clinical notes follow a predictable structure. Physicians write under section headers like "History of Present Illness" and "Physical Examination" that separate different types of clinical information.
Your pipeline needs to split a raw note into these labeled sections before it can extract findings. In this step, you'll build a SectionDetector that uses regex patterns to identify section boundaries and return structured results.
In this step, get ready to:
- Define regex patterns that detect five common clinical note section headers.
- Implement a SectionDetector class that splits a note into labeled sections.
- Test the detector against a multi-section clinical note.
Define regex patterns for clinical section headers
Clinical notes use various header formats. A physician might write "History of Present Illness:", "HPI -", or "PHYSICAL EXAMINATION" to start a section. Your regex patterns need to handle these variations while mapping them to consistent labels.
- Open section_detector.py in VS Code and add the module header and pattern definitions:
"""Clinical note section detection using regex patterns."""
import re
from models import SectionResult
# Regex patterns mapping clinical section headers to standard labels
SECTION_PATTERNS = [
(r"(?i)(history\s+of\s+present\s+illness|hpi)\s*[:\-]?\s*", "HPI"),
(r"(?i)(physical\s+exam(?:ination)?|pe)\s*[:\-]?\s*", "PHYSICAL_EXAM"),
(r"(?i)(chest\s+x[- ]?ray|cxr|imaging|radiology)\s*[:\-]?\s*", "IMAGING"),
(r"(?i)(assessment(?:\s+and\s+plan)?|a/?p)\s*[:\-]?\s*", "ASSESSMENT"),
(r"(?i)(review\s+of\s+systems|ros)\s*[:\-]?\s*", "ROS"),
]What does this code do?
Each tuple in SECTION_PATTERNS pairs a regex pattern with a standard label. The patterns use (?i) for case-insensitive matching so "HPI" and "History of Present Illness" both map to the label "HPI".
The trailing \s*[:\-]?\s* portion matches optional colons, dashes, or spaces that physicians use after headers. This lets the detector handle "HPI:", "HPI -", or just "HPI" followed by a newline.
- Save the file. You won't see a change yet because the class that uses these patterns comes next.
Implement the SectionDetector class
The detector needs to find every section header in a note, determine where each section's content starts and ends, and return structured results. It does this by collecting all header matches, sorting them by position, then slicing the text between consecutive headers.
- Add the SectionDetector class below the SECTION_PATTERNS list in section_detector.py:
class SectionDetector:
"""Detects and splits clinical note sections."""
def __init__(self, patterns=None):
# Use provided patterns or fall back to the module-level defaults
self.patterns = patterns or SECTION_PATTERNS
self._compiled = [(re.compile(p), name) for p, name in self.patterns]
def detect(self, note: str) -> list[SectionResult]:
"""Detect sections in a clinical note, ordered by position."""
# Find all section header matches across all patterns
matches = []
for pattern, section_name in self._compiled:
for match in pattern.finditer(note):
matches.append((match.start(), match.end(), section_name))
# Sort by position so we can slice between consecutive headers
matches.sort(key=lambda x: x[0])
# Extract content between each header and the next
sections = []
for i, (start, header_end, name) in enumerate(matches):
# Content runs from end of this header to start of the next
content_end = matches[i + 1][0] if i + 1 < len(matches) else len(note)
content = note[header_end:content_end].strip()
sections.append(SectionResult(
name=name,
content=content,
start_pos=start,
end_pos=content_end,
))
return sectionsWhat does this code do?
- __init__ pre-compiles all regex patterns once at instantiation. This avoids recompiling on every call to detect().
- detect() uses finditer() to scan the entire note for each pattern. It collects every match with its start position, end position, and section label.
- After sorting by position, it slices the note between consecutive headers. The last section extends to the end of the note.
- Each slice is returned as a SectionResult dataclass (which you already defined in models.py).
- Save section_detector.py.
Seeing a syntax error on save?
Check that your SECTION_PATTERNS list ends with a closing ] before the class definition. Also make sure each regex string uses a raw string prefix (r"...") so backslashes aren't interpreted as escape characters.
Help me debug a syntax error in my section_detector.py regex patterns.
Test with a synthetic clinical note
Before moving on, you should verify that the detector correctly identifies sections from a realistic clinical note with multiple headers and varying formats.
- Test the section detector by running this command from your project folder:
python -c "
from section_detector import SectionDetector
note = '''History of Present Illness:
Patient has cough and fever for 3 days.
Physical Examination:
Bilateral wheezing noted. No crackles.
Chest X-Ray:
Right lower lobe consolidation.'''
detector = SectionDetector()
sections = detector.detect(note)
for s in sections:
print(f'{s.name}: {s.content[:50]}')
"What should I see?
You should see three labeled sections printed:
HPI: Patient has cough and fever for 3 days.
PHYSICAL_EXAM: Bilateral wheezing noted. No crackles.
IMAGING: Right lower lobe consolidation.Each section header was matched, its content was extracted, and the correct label was assigned.
Seeing a ModuleNotFoundError?
Make sure you're running the command from inside your project folder (where section_detector.py and models.py live). Python needs both files in the same directory to resolve the import.
Help me fix a ModuleNotFoundError when running my section detector test.
✔️ Awesome, I've got everything!
Great. Double check you've saved section_detector.py and your three sections printed correctly.
ⓧ I'd like to double check the full code
Here is the complete section_detector.py file:
"""Clinical note section detection using regex patterns."""
import re
from models import SectionResult
# Regex patterns mapping clinical section headers to standard labels
SECTION_PATTERNS = [
(r"(?i)(history\s+of\s+present\s+illness|hpi)\s*[:\-]?\s*", "HPI"),
(r"(?i)(physical\s+exam(?:ination)?|pe)\s*[:\-]?\s*", "PHYSICAL_EXAM"),
(r"(?i)(chest\s+x[- ]?ray|cxr|imaging|radiology)\s*[:\-]?\s*", "IMAGING"),
(r"(?i)(assessment(?:\s+and\s+plan)?|a/?p)\s*[:\-]?\s*", "ASSESSMENT"),
(r"(?i)(review\s+of\s+systems|ros)\s*[:\-]?\s*", "ROS"),
]
class SectionDetector:
"""Detects and splits clinical note sections."""
def __init__(self, patterns=None):
# Use provided patterns or fall back to the module-level defaults
self.patterns = patterns or SECTION_PATTERNS
self._compiled = [(re.compile(p), name) for p, name in self.patterns]
def detect(self, note: str) -> list[SectionResult]:
"""Detect sections in a clinical note, ordered by position."""
# Find all section header matches across all patterns
matches = []
for pattern, section_name in self._compiled:
for match in pattern.finditer(note):
matches.append((match.start(), match.end(), section_name))
# Sort by position so we can slice between consecutive headers
matches.sort(key=lambda x: x[0])
# Extract content between each header and the next
sections = []
for i, (start, header_end, name) in enumerate(matches):
# Content runs from end of this header to start of the next
content_end = matches[i + 1][0] if i + 1 < len(matches) else len(note)
content = note[header_end:content_end].strip()
sections.append(SectionResult(
name=name,
content=content,
start_pos=start,
end_pos=content_end,
))
return sectionsYour section detector can now split any clinical note into labeled chunks. Next up, you'll build the finding extractor that scans these sections for respiratory symptoms like wheezing, crackles, and consolidation.
Build the Finding Extractor and Negation Detector
Your section detector can now split a clinical note into labeled regions. The next challenge is extracting specific medical findings from those sections.
Physicians write in shorthand and natural language, so "wheezing" might appear as "wheezes", "wheezy", or "bilateral wheeze." On top of that, a finding like "no wheezing" means the opposite of "wheezing noted." In this step, you'll build two modules that handle both problems: a finding extractor that catches clinical language variations, and a negation detector that distinguishes affirmed findings from negated ones.
In this step, get ready to:
- Build a finding extractor with 9 regex patterns covering respiratory symptoms.
- Build a negation detector that uses scope windows and clinical cue patterns.
- Verify both modules work together to correctly classify affirmed and negated findings.
Define finding patterns and the extraction data model
Clinical text uses many synonyms for the same finding. "Crackles" might be written as "rales" or "crepitations." Each regex pattern needs to capture these variations so the extractor doesn't miss relevant mentions.
- Open finding_extractor.py in VS Code and add the imports, pattern definitions, and data model by pasting this code:
"""Respiratory finding extraction using regex patterns."""
import re
from dataclasses import dataclass
from models import SectionResult
# Respiratory finding patterns with clinical variations
FINDING_PATTERNS = {
"wheezing": r"(?i)\b(wheez(?:ing|es|e)?|wheezy)\b",
"crackles": r"(?i)\b(crackles?|rales?|crepitation(?:s)?)\b",
"rhonchi": r"(?i)\b(rhonchi|coarse\s+breath\s+sounds?)\b",
"consolidation": r"(?i)\b(consolidat(?:ion|ed|ing))\b",
"effusion": r"(?i)\b((?:pleural\s+)?effusion)\b",
"decreased_breath_sounds": r"(?i)\b(decreased\s+breath\s+sounds?|diminished\s+breath\s+sounds?)\b",
"fever": r"(?i)\b(fever|febrile|temp(?:erature)?\s+(?:of\s+)?(?:10[0-4]|39|40))\b",
"dyspnea": r"(?i)\b(dyspn[eo]a|shortness\s+of\s+breath|sob|breathless(?:ness)?)\b",
"cough": r"(?i)\b(cough(?:ing)?|productive\s+cough)\b",
}
@dataclass
class RawFindingMatch:
"""Intermediate finding match before negation processing."""
name: str
text_span: str
start_pos: int
end_pos: int
section_name: strWhat does this code do?
FINDING_PATTERNS is a dictionary mapping each canonical finding name to a regex that captures its clinical variations. For example, the "crackles" pattern also matches "rales" and "crepitations" because physicians use all three terms interchangeably.
RawFindingMatch is a lightweight dataclass that holds an intermediate extraction result before negation is applied. It records the finding name, exact matched text, character positions, and which section it came from.
- Save finding_extractor.py.
You won't see a change yet. The next chunk adds the class that uses these patterns.
Implement the FindingExtractor class
The extractor iterates over every section from the section detector, applies each finding pattern, and records every match with its exact position in the original note.
- In finding_extractor.py, add the FindingExtractor class below the RawFindingMatch dataclass:
class FindingExtractor:
"""Extracts respiratory findings from clinical text."""
def __init__(self, patterns: dict[str, str] = None):
self.patterns = patterns or FINDING_PATTERNS
self._compiled = {
name: re.compile(pattern)
for name, pattern in self.patterns.items()
}
def extract(self, sections: list[SectionResult]) -> list[RawFindingMatch]:
"""Extract findings from all sections.
Returns raw matches (negation not yet applied).
"""
matches = []
for section in sections:
for finding_name, pattern in self._compiled.items():
for match in pattern.finditer(section.content):
matches.append(RawFindingMatch(
name=finding_name,
text_span=match.group(0),
start_pos=section.start_pos + match.start(),
end_pos=section.start_pos + match.end(),
section_name=section.name,
))
return matchesWhat does this code do?
The constructor pre-compiles all regex patterns for performance. The extract() method loops through each section, applies every finding pattern, and records each hit as a RawFindingMatch.
Notice the position math: section.start_pos + match.start() translates the match position within the section content back to a position in the full original note. This is critical for the negation detector, which needs to look at surrounding context in the original text.
- Save finding_extractor.py.
- Verify the extractor works by running this quick test in your terminal:
python3 -c "from finding_extractor import FindingExtractor; from models import SectionResult; e = FindingExtractor(); results = e.extract([SectionResult(name='PE', content='Bilateral wheezing and crackles noted.', start_pos=0, end_pos=38)]); print([(r.name, r.text_span) for r in results])"You should see output containing tuples like ('wheezing', 'wheezing') and ('crackles', 'crackles').
Seeing a ModuleNotFoundError?
Make sure you're running the command from inside your respiratory_pipeline project folder. Both models.py and finding_extractor.py need to be in the same directory.
help me debug a ModuleNotFoundError when importing finding_extractor
✔️ Awesome, I've got everything!
Great. Double check you've saved finding_extractor.py before moving on.
ⓧ I'd like to double check the full code
"""Respiratory finding extraction using regex patterns."""
import re
from dataclasses import dataclass
from models import SectionResult
FINDING_PATTERNS = {
"wheezing": r"(?i)\b(wheez(?:ing|es|e)?|wheezy)\b",
"crackles": r"(?i)\b(crackles?|rales?|crepitation(?:s)?)\b",
"rhonchi": r"(?i)\b(rhonchi|coarse\s+breath\s+sounds?)\b",
"consolidation": r"(?i)\b(consolidat(?:ion|ed|ing))\b",
"effusion": r"(?i)\b((?:pleural\s+)?effusion)\b",
"decreased_breath_sounds": r"(?i)\b(decreased\s+breath\s+sounds?|diminished\s+breath\s+sounds?)\b",
"fever": r"(?i)\b(fever|febrile|temp(?:erature)?\s+(?:of\s+)?(?:10[0-4]|39|40))\b",
"dyspnea": r"(?i)\b(dyspn[eo]a|shortness\s+of\s+breath|sob|breathless(?:ness)?)\b",
"cough": r"(?i)\b(cough(?:ing)?|productive\s+cough)\b",
}
@dataclass
class RawFindingMatch:
name: str
text_span: str
start_pos: int
end_pos: int
section_name: str
class FindingExtractor:
def __init__(self, patterns=None):
self.patterns = patterns or FINDING_PATTERNS
self._compiled = {name: re.compile(pattern) for name, pattern in self.patterns.items()}
def extract(self, sections: list[SectionResult]) -> list[RawFindingMatch]:
matches = []
for section in sections:
for finding_name, pattern in self._compiled.items():
for match in pattern.finditer(section.content):
matches.append(RawFindingMatch(
name=finding_name,
text_span=match.group(0),
start_pos=section.start_pos + match.start(),
end_pos=section.start_pos + match.end(),
section_name=section.name,
))
return matchesDefine negation cues and scope constants
In clinical notes, negation follows predictable patterns. A physician writes "no wheezing" or "denies fever" to indicate an absence. The negation detector works by looking at a small window of text around each extracted finding and checking for these cue words.
What is a negation scope window?
Negation cues only apply to nearby findings. "No wheezing or crackles" negates both, but "No fever. Wheezing present." only negates fever. The scope window (30 characters before, 15 after) defines how far a negation cue's influence extends. This is a simplified version of the NegEx algorithm used in production clinical NLP systems.
- Open negation_detector.py in VS Code and add the imports, negation cue lists, and scope constants by pasting this code:
"""Clinical negation detection for extracted findings."""
import re
from finding_extractor import RawFindingMatch
from models import ClinicalFinding
# Negation cue patterns (pre-finding negation)
PRE_NEGATION_CUES = [
r"(?i)\bno\b",
r"(?i)\bdenies\b",
r"(?i)\bdenied\b",
r"(?i)\bwithout\b",
r"(?i)\bnegative\s+for\b",
r"(?i)\babsent\b",
r"(?i)\bnot\b",
r"(?i)\brules?\s+out\b",
r"(?i)\bno\s+evidence\s+of\b",
]
# Post-finding negation (less common but occurs in clinical text)
POST_NEGATION_CUES = [
r"(?i)\babsent\b",
r"(?i)\bnot\s+(?:seen|noted|observed|present)\b",
]
# Default scope: how many characters before/after a finding to look for negation
DEFAULT_PRE_SCOPE = 30
DEFAULT_POST_SCOPE = 15What does this code do?
PRE_NEGATION_CUES lists 9 regex patterns for words that negate a following finding ("no", "denies", "without", etc.). POST_NEGATION_CUES lists 2 patterns for negation that appears after a finding ("absent", "not seen").
DEFAULT_PRE_SCOPE = 30 means the detector looks up to 30 characters before a finding for negation cues. DEFAULT_POST_SCOPE = 15 looks 15 characters after. These values balance catching real negations without reaching too far into unrelated text.
- Save negation_detector.py.
You won't see a change yet. The next chunk adds the class that applies these cues to extracted findings.
Implement the NegationDetector class
The detector takes raw finding matches and the full note text. For each finding, it checks a window of surrounding characters for negation cues and produces a final ClinicalFinding with the is_negated flag set correctly.
- In negation_detector.py, add the NegationDetector class below the scope constants:
class NegationDetector:
"""Detects negation for clinical findings."""
def __init__(
self,
pre_cues: list[str] = None,
post_cues: list[str] = None,
pre_scope: int = DEFAULT_PRE_SCOPE,
post_scope: int = DEFAULT_POST_SCOPE,
):
self.pre_cues = [re.compile(c) for c in (pre_cues or PRE_NEGATION_CUES)]
self.post_cues = [re.compile(c) for c in (post_cues or POST_NEGATION_CUES)]
self.pre_scope = pre_scope
self.post_scope = post_scope
def apply(
self, raw_matches: list[RawFindingMatch], full_text: str
) -> list[ClinicalFinding]:
"""Apply negation detection to raw finding matches.
Returns ClinicalFinding objects with is_negated set.
"""
findings = []
for match in raw_matches:
is_negated = self._check_negation(match, full_text)
findings.append(ClinicalFinding(
name=match.name,
section=match.section_name,
is_negated=is_negated,
text_span=match.text_span,
start_pos=match.start_pos,
end_pos=match.end_pos,
))
return findings
def _check_negation(self, match: RawFindingMatch, full_text: str) -> bool:
"""Check if a finding is negated by looking at surrounding context."""
# Check pre-negation (text before the finding)
pre_start = max(0, match.start_pos - self.pre_scope)
pre_text = full_text[pre_start:match.start_pos]
for cue in self.pre_cues:
if cue.search(pre_text):
return True
# Check post-negation (text after the finding)
post_end = min(len(full_text), match.end_pos + self.post_scope)
post_text = full_text[match.end_pos:post_end]
for cue in self.post_cues:
if cue.search(post_text):
return True
return FalseWhat does this code do?
The apply() method converts each RawFindingMatch into a final ClinicalFinding dataclass (from models.py) with the is_negated boolean set by _check_negation().
_check_negation() slices a window of text before and after the finding's position. It searches the pre-window (30 chars) for cues like "no" or "denies", then the post-window (15 chars) for cues like "absent." If any cue matches, the finding is marked as negated.
- Save negation_detector.py.
- Verify both modules work together by running this end-to-end test in your terminal:
python3 -c "
from models import SectionResult
from finding_extractor import FindingExtractor
from negation_detector import NegationDetector
note = 'Physical Examination: Bilateral wheezing. No crackles. Denies fever.'
sections = [SectionResult(name='PE', content='Bilateral wheezing. No crackles. Denies fever.', start_pos=22, end_pos=68)]
extractor = FindingExtractor()
raw = extractor.extract(sections)
detector = NegationDetector()
findings = detector.apply(raw, note)
for f in findings:
status = 'NEGATED' if f.is_negated else 'AFFIRMED'
print(f' {f.name}: {status}')
"You should see output showing wheezing: AFFIRMED, crackles: NEGATED, and fever: NEGATED.
Seeing all findings marked AFFIRMED?
Check that your start_pos value in the test's SectionResult matches where the section content starts in the full note string. The negation detector uses these positions to slice the full note text for context. If positions are wrong, it looks at the wrong characters and misses negation cues.
help me debug why my NegationDetector marks everything as affirmed
✔️ Awesome, I've got everything!
Great. Double check you've saved negation_detector.py before moving on.
ⓧ I'd like to double check the full code
"""Clinical negation detection for extracted findings."""
import re
from finding_extractor import RawFindingMatch
from models import ClinicalFinding
PRE_NEGATION_CUES = [
r"(?i)\bno\b",
r"(?i)\bdenies\b",
r"(?i)\bdenied\b",
r"(?i)\bwithout\b",
r"(?i)\bnegative\s+for\b",
r"(?i)\babsent\b",
r"(?i)\bnot\b",
r"(?i)\brules?\s+out\b",
r"(?i)\bno\s+evidence\s+of\b",
]
POST_NEGATION_CUES = [
r"(?i)\babsent\b",
r"(?i)\bnot\s+(?:seen|noted|observed|present)\b",
]
DEFAULT_PRE_SCOPE = 30
DEFAULT_POST_SCOPE = 15
class NegationDetector:
def __init__(self, pre_cues=None, post_cues=None, pre_scope=DEFAULT_PRE_SCOPE, post_scope=DEFAULT_POST_SCOPE):
self.pre_cues = [re.compile(c) for c in (pre_cues or PRE_NEGATION_CUES)]
self.post_cues = [re.compile(c) for c in (post_cues or POST_NEGATION_CUES)]
self.pre_scope = pre_scope
self.post_scope = post_scope
def apply(self, raw_matches: list[RawFindingMatch], full_text: str) -> list[ClinicalFinding]:
findings = []
for match in raw_matches:
is_negated = self._check_negation(match, full_text)
findings.append(ClinicalFinding(
name=match.name,
section=match.section_name,
is_negated=is_negated,
text_span=match.text_span,
start_pos=match.start_pos,
end_pos=match.end_pos,
))
return findings
def _check_negation(self, match: RawFindingMatch, full_text: str) -> bool:
pre_start = max(0, match.start_pos - self.pre_scope)
pre_text = full_text[pre_start:match.start_pos]
for cue in self.pre_cues:
if cue.search(pre_text):
return True
post_end = min(len(full_text), match.end_pos + self.post_scope)
post_text = full_text[match.end_pos:post_end]
for cue in self.post_cues:
if cue.search(post_text):
return True
return FalseYour extraction layer is complete. The finding extractor catches 9 types of respiratory symptoms regardless of how the physician phrases them. The negation detector correctly flips the meaning when cue words like "no" or "denies" appear nearby. Next up, you'll wire these modules into a pipeline orchestrator that chains all processing stages together automatically.
Wire Up the Pipeline Orchestrator
You have all four NLP modules built: section detection, finding extraction, negation handling, and typed dataclass models. Right now they sit in separate files with no connection between them.
In this step, you will create a pipeline orchestrator that chains all three processing stages into one call. You will also build a set of synthetic clinical notes that give the pipeline realistic text to process.
In this step, get ready to:
- Create a pipeline class that chains section detection, finding extraction, and negation handling.
- Build synthetic clinical notes with ground-truth disease labels.
- Run the full NLP pipeline end-to-end on a test note.
Create the pipeline orchestrator class
The orchestrator's job is simple: take a raw clinical note, pass it through each NLP stage in order, and collect the results into a single PipelineOutput object. If any stage throws an exception, the orchestrator catches it, logs the error, and returns what it has so far.
- Open pipeline.py in VS Code and add the following code:
"""Pipeline orchestrator that chains all processing stages."""
from models import PipelineOutput
from section_detector import SectionDetector
from finding_extractor import FindingExtractor
from negation_detector import NegationDetector
class ClinicalPipeline:
"""Orchestrates the clinical NLP processing pipeline."""
def __init__(self):
self.section_detector = SectionDetector()
self.finding_extractor = FindingExtractor()
self.negation_detector = NegationDetector()
def process(self, note: str) -> PipelineOutput:
"""Process a clinical note through all pipeline stages.
Pipeline flow:
Raw Note -> Section Detection -> Finding Extraction
-> Negation Handling -> Structured Output
"""
output = PipelineOutput(raw_note=note)
# Stage 1: Section Detection
try:
output.sections = self.section_detector.detect(note)
except Exception as e:
output.errors.append(f"Section detection failed: {e}")
return output
# Stage 2: Finding Extraction
try:
raw_matches = self.finding_extractor.extract(output.sections)
except Exception as e:
output.errors.append(f"Finding extraction failed: {e}")
return output
# Stage 3: Negation Detection
try:
output.findings = self.negation_detector.apply(raw_matches, note)
except Exception as e:
output.errors.append(f"Negation detection failed: {e}")
return output
return outputWhat does this code do?
ClinicalPipeline instantiates all three NLP stages in its constructor. The process() method chains them in sequence: sections first, then findings from those sections, then negation applied to those findings.
Each stage is wrapped in a try/except block. If a stage fails, the error message is appended to output.errors and the pipeline returns immediately with whatever it has computed so far. This prevents a crash in one stage from losing all upstream results.
- Save pipeline.py.
- Verify the import chain works by running this in your terminal:
python -c "from pipeline import ClinicalPipeline; print('Pipeline imported successfully')"You should see Pipeline imported successfully printed with no errors.
Seeing an ImportError?
Make sure your terminal is inside the respiratory_pipeline project folder. All module files (models.py, section_detector.py, finding_extractor.py, negation_detector.py) must be in the same directory as pipeline.py.
If you see ModuleNotFoundError: No module named 'models', confirm you are running the command from the correct folder. Help me fix this ImportError in my pipeline project.
✔️ Awesome, I've got everything!
Great. Make sure you have saved pipeline.py before moving on.
ⓧ I'd like to double check the full code
"""Pipeline orchestrator that chains all processing stages."""
from models import PipelineOutput
from section_detector import SectionDetector
from finding_extractor import FindingExtractor
from negation_detector import NegationDetector
class ClinicalPipeline:
"""Orchestrates the clinical NLP processing pipeline."""
def __init__(self):
self.section_detector = SectionDetector()
self.finding_extractor = FindingExtractor()
self.negation_detector = NegationDetector()
def process(self, note: str) -> PipelineOutput:
"""Process a clinical note through all pipeline stages.
Pipeline flow:
Raw Note -> Section Detection -> Finding Extraction
-> Negation Handling -> Structured Output
"""
output = PipelineOutput(raw_note=note)
# Stage 1: Section Detection
try:
output.sections = self.section_detector.detect(note)
except Exception as e:
output.errors.append(f"Section detection failed: {e}")
return output
# Stage 2: Finding Extraction
try:
raw_matches = self.finding_extractor.extract(output.sections)
except Exception as e:
output.errors.append(f"Finding extraction failed: {e}")
return output
# Stage 3: Negation Detection
try:
output.findings = self.negation_detector.apply(raw_matches, note)
except Exception as e:
output.errors.append(f"Negation detection failed: {e}")
return output
return outputCreate synthetic clinical notes for testing
The pipeline needs clinical text to process. Real patient data is protected, so you will build a set of synthetic notes that mimic real clinical formatting.
- Open synthetic_data.py in VS Code and add the following code:
"""Synthetic clinical notes for testing and training."""
SYNTHETIC_NOTES = [
{
"note": """History of Present Illness:
65-year-old male presents with 3 days of productive cough, fever, and shortness of breath. Patient reports rust-colored sputum.
Physical Examination:
Lungs: Crackles noted in right lower lobe. No wheezing. Decreased breath sounds at right base.
Vitals: Temp 101.8F, RR 24, SpO2 92% on room air.
Chest X-Ray:
Right lower lobe consolidation with small pleural effusion.
Assessment:
Community-acquired pneumonia.""",
"disease": "pneumonia",
},
{
"note": """HPI:
72-year-old female with history of COPD presents with worsening dyspnea and increased sputum production over 5 days. Denies fever. Reports wheezing worse in the morning.
PE:
Bilateral wheezing and rhonchi. No crackles. No consolidation on percussion.
Imaging:
Hyperinflated lungs. No consolidation. No effusion.
Assessment and Plan:
COPD exacerbation, start steroids and bronchodilators.""",
"disease": "copd exacerbation",
},
{
"note": """History of Present Illness:
28-year-old female presents to ED with acute shortness of breath and wheezing that started 2 hours ago after exposure to cat dander. Denies fever. Denies productive cough.
Physical Examination:
Diffuse bilateral wheezing. No crackles. No rhonchi. Good air movement.
Chest X-Ray:
Clear lungs. No consolidation. No effusion.
Assessment:
Acute asthma exacerbation triggered by allergen exposure.""",
"disease": "asthma attack",
},
{
"note": """HPI:
78-year-old male with history of CHF presents with progressive dyspnea on exertion and orthopnea over 1 week. Denies fever. Reports bilateral lower extremity edema.
PE:
Bilateral crackles at bases. No wheezing. JVD present. 2+ pitting edema bilateral lower extremities.
CXR:
Bilateral pleural effusions. Pulmonary edema. No consolidation.
A/P:
Decompensated heart failure. Diuresis and afterload reduction.""",
"disease": "heart failure",
},
{
"note": """History of Present Illness:
55-year-old male smoker with 5 days of worsening cough and fever. Shortness of breath at rest. Reports green sputum.
Physical Examination:
Temperature 102.4F. Crackles in left lower lobe. Decreased breath sounds left base. No wheezing.
Chest X-Ray:
Left lower lobe consolidation. Small left-sided effusion.
Assessment:
Pneumonia, left lower lobe.""",
"disease": "pneumonia",
},
]
def get_training_notes() -> list[dict]:
"""Return synthetic notes for training CPTs."""
return SYNTHETIC_NOTES
def get_test_note() -> str:
"""Return a single test note for pipeline demonstration."""
return SYNTHETIC_NOTES[0]["note"]What does this code do?
SYNTHETIC_NOTES is a list of 5 annotated clinical notes. Each entry has a note field (the raw physician text) and a disease field (the ground-truth diagnosis).
The notes cover four respiratory conditions: pneumonia (2 cases), COPD exacerbation, asthma attack, and heart failure. Each note follows real clinical formatting with HPI, Physical Exam, Imaging, and Assessment sections.
get_training_notes() returns all 5 notes for building probability tables later. get_test_note() returns just the first note for quick testing.
- Save synthetic_data.py.
- Verify the file loads correctly by running:
python -c "from synthetic_data import get_training_notes; print(f'{len(get_training_notes())} notes loaded')"You should see 5 notes loaded printed.
Seeing a SyntaxError?
Triple-quoted strings ("""...""") are sensitive to indentation. Make sure the closing """ for each note sits on the same line as the last line of note text, not on a new line with extra whitespace.
Help me debug a SyntaxError in my synthetic_data.py file.
✔️ Awesome, I've got everything!
Great. Make sure you have saved synthetic_data.py before moving on.
ⓧ I'd like to double check the full code
"""Synthetic clinical notes for testing and training."""
SYNTHETIC_NOTES = [
{
"note": """History of Present Illness:
65-year-old male presents with 3 days of productive cough, fever, and shortness of breath. Patient reports rust-colored sputum.
Physical Examination:
Lungs: Crackles noted in right lower lobe. No wheezing. Decreased breath sounds at right base.
Vitals: Temp 101.8F, RR 24, SpO2 92% on room air.
Chest X-Ray:
Right lower lobe consolidation with small pleural effusion.
Assessment:
Community-acquired pneumonia.""",
"disease": "pneumonia",
},
{
"note": """HPI:
72-year-old female with history of COPD presents with worsening dyspnea and increased sputum production over 5 days. Denies fever. Reports wheezing worse in the morning.
PE:
Bilateral wheezing and rhonchi. No crackles. No consolidation on percussion.
Imaging:
Hyperinflated lungs. No consolidation. No effusion.
Assessment and Plan:
COPD exacerbation, start steroids and bronchodilators.""",
"disease": "copd exacerbation",
},
{
"note": """History of Present Illness:
28-year-old female presents to ED with acute shortness of breath and wheezing that started 2 hours ago after exposure to cat dander. Denies fever. Denies productive cough.
Physical Examination:
Diffuse bilateral wheezing. No crackles. No rhonchi. Good air movement.
Chest X-Ray:
Clear lungs. No consolidation. No effusion.
Assessment:
Acute asthma exacerbation triggered by allergen exposure.""",
"disease": "asthma attack",
},
{
"note": """HPI:
78-year-old male with history of CHF presents with progressive dyspnea on exertion and orthopnea over 1 week. Denies fever. Reports bilateral lower extremity edema.
PE:
Bilateral crackles at bases. No wheezing. JVD present. 2+ pitting edema bilateral lower extremities.
CXR:
Bilateral pleural effusions. Pulmonary edema. No consolidation.
A/P:
Decompensated heart failure. Diuresis and afterload reduction.""",
"disease": "heart failure",
},
{
"note": """History of Present Illness:
55-year-old male smoker with 5 days of worsening cough and fever. Shortness of breath at rest. Reports green sputum.
Physical Examination:
Temperature 102.4F. Crackles in left lower lobe. Decreased breath sounds left base. No wheezing.
Chest X-Ray:
Left lower lobe consolidation. Small left-sided effusion.
Assessment:
Pneumonia, left lower lobe.""",
"disease": "pneumonia",
},
]
def get_training_notes() -> list[dict]:
"""Return synthetic notes for training CPTs."""
return SYNTHETIC_NOTES
def get_test_note() -> str:
"""Return a single test note for pipeline demonstration."""
return SYNTHETIC_NOTES[0]["note"]Run the pipeline end-to-end on a test note
Time to see if all three stages actually chain together. You will feed a synthetic pneumonia note into the pipeline and inspect what comes out the other side.
- Run the full pipeline on the first synthetic note by running this command:
python -c "from pipeline import ClinicalPipeline; from synthetic_data import get_test_note; o = ClinicalPipeline().process(get_test_note()); print(f'Sections: {len(o.sections)}'); print(f'Findings: {len(o.findings)}'); print(f'Affirmed: {len(o.get_affirmed_findings())}'); print(f'Negated: {len(o.get_negated_findings())}')"What should I see?
You should see output showing multiple sections detected, several findings extracted, and at least one negated finding. The pneumonia note contains "No wheezing" which should be classified as negated, while "Crackles" and "consolidation" should be affirmed.
The exact counts depend on how the regex patterns match the clinical text. The key confirmation is that all four numbers are greater than zero (sections, findings, affirmed, and negated each have at least one entry).
All counts showing zero?
If sections is 0, the section detector regex patterns are not matching the note headers. Check that section_detector.py has the correct SECTION_PATTERNS list.
If findings is 0 but sections is not, the finding extractor is not matching within the section content. Confirm that finding_extractor.py has the FINDING_PATTERNS dictionary with patterns for wheezing, crackles, and consolidation.
Help me debug why my pipeline is returning zero findings.
Your pipeline orchestrator is wired up and processing clinical notes end-to-end. Next up, you will write a comprehensive test suite to validate each stage independently.
Validate the Pipeline with Unit Tests
You now have a complete NLP pipeline that detects sections, extracts findings, and handles negation. But how do you know each stage actually works correctly on edge cases?
In this step, you will write a pytest test suite that validates each pipeline stage independently and as an integrated whole. Testing each stage in isolation catches bugs at the source instead of letting them cascade downstream.
In this step, get ready to:
- Write tests for section detection covering correct boundaries, missing sections, and edge cases.
- Write tests for finding extraction validating known findings and rejecting unknown text.
- Write tests for negation handling and full pipeline integration.
Write tests for section detection
The section detector is the first stage of the pipeline. If it fails to find section boundaries, every downstream stage receives garbage input. Your tests need to cover the happy path, missing headers, and case variations.
- Open test_pipeline.py in VS Code.
- Add the imports and the TestSectionDetector class by pasting this code:
"""Unit tests for the clinical NLP pipeline."""
import pytest
from models import SectionResult
from section_detector import SectionDetector
from finding_extractor import FindingExtractor, RawFindingMatch
from negation_detector import NegationDetector
from pipeline import ClinicalPipeline
class TestSectionDetector:
def setup_method(self):
self.detector = SectionDetector()
def test_detects_hpi_section(self):
note = "History of Present Illness:\nPatient has cough."
sections = self.detector.detect(note)
assert len(sections) >= 1
assert sections[0].name == "HPI"
assert "cough" in sections[0].content
def test_detects_multiple_sections(self):
note = "HPI: Patient has fever.\nPhysical Examination: Lungs clear.\nChest X-Ray: Normal."
sections = self.detector.detect(note)
assert len(sections) == 3
assert sections[0].name == "HPI"
assert sections[1].name == "PHYSICAL_EXAM"
assert sections[2].name == "IMAGING"
def test_handles_no_sections(self):
note = "This is just plain text without section headers."
sections = self.detector.detect(note)
assert len(sections) == 0
def test_handles_case_insensitive_headers(self):
note = "PHYSICAL EXAMINATION:\nWheezing noted."
sections = self.detector.detect(note)
assert len(sections) == 1
assert sections[0].name == "PHYSICAL_EXAM"What does this code do?
The imports pull in every class your tests need to exercise. pytest discovers test classes and methods automatically by name.
Each test in TestSectionDetector validates one scenario: detecting a single section, detecting multiple sections, handling notes with no section headers, and matching headers regardless of case. The setup_method creates a fresh detector instance before each test so they stay independent.
- Save test_pipeline.py.
Seeing an import error?
Make sure your terminal is inside the respiratory_pipeline folder where all your module files live. Pytest needs to find models.py, section_detector.py, and the other modules in the same directory.
help me debug import errors in my pytest test file
Write tests for finding extraction
The finding extractor needs to reliably pick up respiratory terms from clinical text while ignoring unrelated prose. These tests check both true positives and true negatives.
- In test_pipeline.py, add the TestFindingExtractor class below the TestSectionDetector class:
class TestFindingExtractor:
def setup_method(self):
self.extractor = FindingExtractor()
def test_extracts_wheezing(self):
sections = [SectionResult(name="PHYSICAL_EXAM", content="Bilateral wheezing noted.", start_pos=0, end_pos=25)]
matches = self.extractor.extract(sections)
assert "wheezing" in [m.name for m in matches]
def test_extracts_crackles(self):
sections = [SectionResult(name="PHYSICAL_EXAM", content="Crackles in right lower lobe.", start_pos=0, end_pos=30)]
matches = self.extractor.extract(sections)
assert "crackles" in [m.name for m in matches]
def test_extracts_consolidation_from_imaging(self):
sections = [SectionResult(name="IMAGING", content="Right lower lobe consolidation.", start_pos=0, end_pos=32)]
matches = self.extractor.extract(sections)
assert "consolidation" in [m.name for m in matches]
def test_no_false_positives_on_clean_text(self):
sections = [SectionResult(name="HPI", content="Patient feels well today.", start_pos=0, end_pos=25)]
matches = self.extractor.extract(sections)
assert len(matches) == 0What does this code do?
Each test creates a SectionResult with controlled clinical text, then verifies the extractor either finds the expected finding or correctly produces zero matches. The test_no_false_positives_on_clean_text test is especially important. It confirms the extractor does not hallucinate findings from normal English.
- Save test_pipeline.py.
Test failing on crackles or consolidation?
Double check that your finding_extractor.py has entries for both crackles and consolidation in the FINDING_PATTERNS dictionary. A missing or misspelled key will cause the extractor to skip that finding entirely.
help me debug my finding extractor tests
Write tests for negation handling and integration
Negation is where clinical NLP gets tricky. "No wheezing" and "Bilateral wheezing" contain the same word but mean opposite things. These tests verify your negation detector correctly classifies both cases.
- In test_pipeline.py, add the TestNegationDetector class below TestFindingExtractor:
class TestNegationDetector:
def setup_method(self):
self.detector = NegationDetector()
def test_detects_no_wheezing(self):
match = RawFindingMatch(name="wheezing", text_span="wheezing", start_pos=3, end_pos=11, section_name="PHYSICAL_EXAM")
findings = self.detector.apply([match], "No wheezing noted.")
assert findings[0].is_negated is True
def test_affirmed_finding(self):
match = RawFindingMatch(name="wheezing", text_span="wheezing", start_pos=11, end_pos=19, section_name="PHYSICAL_EXAM")
findings = self.detector.apply([match], "Bilateral wheezing noted.")
assert findings[0].is_negated is False
def test_denies_negation(self):
match = RawFindingMatch(name="fever", text_span="fever", start_pos=8, end_pos=13, section_name="HPI")
findings = self.detector.apply([match], "Denies fever or chills.")
assert findings[0].is_negated is TrueWhat does this code do?
Each test manually constructs a RawFindingMatch with precise character positions, then passes it alongside the full text to the negation detector. The start_pos and end_pos must match exactly where the finding appears in the string so the scope window looks at the right surrounding context.
For example, in "No wheezing noted.", the word "wheezing" starts at position 3 and ends at position 11. The detector looks back 30 characters and finds "No" within that window, correctly marking it as negated.
- Save test_pipeline.py.
Now add the integration test class that validates all stages working together as a pipeline.
- Below TestNegationDetector, add the TestPipelineIntegration class:
class TestPipelineIntegration:
def setup_method(self):
self.pipeline = ClinicalPipeline()
def test_full_pipeline_pneumonia_note(self):
note = "History of Present Illness:\nPatient presents with fever and cough.\nPhysical Examination:\nCrackles in right lower lobe. No wheezing.\nChest X-Ray:\nRight lower lobe consolidation."
output = self.pipeline.process(note)
assert len(output.errors) == 0
assert len(output.sections) == 3
finding_names = [f.name for f in output.findings]
assert "crackles" in finding_names
assert "wheezing" in finding_names
assert "consolidation" in finding_names
wheezing = [f for f in output.findings if f.name == "wheezing"][0]
assert wheezing.is_negated is True
crackles = [f for f in output.findings if f.name == "crackles"][0]
assert crackles.is_negated is False
def test_pipeline_returns_evidence_dict(self):
note = "Physical Examination:\nBilateral wheezing. No crackles."
output = self.pipeline.process(note)
evidence = output.to_evidence_dict()
assert evidence["Wheezing"] == "present"
assert evidence["Crackles"] == "absent"What does this code do?
The integration tests feed a full clinical note through the entire ClinicalPipeline and verify the end-to-end output. test_full_pipeline_pneumonia_note checks that sections are detected, findings are extracted, and negation is applied correctly in a single pass.
The test_pipeline_returns_evidence_dict test validates the to_evidence_dict() method. This is the bridge between the NLP pipeline and the Bayesian network you will build next. Wheezing (affirmed) maps to "present" and crackles (negated) maps to "absent".
- Save test_pipeline.py.
- Run the full test suite by running this command:
pytest test_pipeline.py -vYou should see all 13 tests pass with green output.
Seeing test failures?
If a negation test fails, check that your start_pos and end_pos values in the test match the character positions in the full text string. The negation scope window depends on these being accurate.
If an integration test fails, verify that pipeline.py correctly imports and chains all three stages. A typo in any import statement will cause the pipeline to fail silently and return errors.
help me debug my failing pytest tests
✔️ Awesome, I've got everything!
Great. Double check you have saved test_pipeline.py and all 13 tests pass.
ⓧ I'd like to double check the full code
"""Unit tests for the clinical NLP pipeline."""
import pytest
from models import SectionResult
from section_detector import SectionDetector
from finding_extractor import FindingExtractor, RawFindingMatch
from negation_detector import NegationDetector
from pipeline import ClinicalPipeline
class TestSectionDetector:
def setup_method(self):
self.detector = SectionDetector()
def test_detects_hpi_section(self):
note = "History of Present Illness:\nPatient has cough."
sections = self.detector.detect(note)
assert len(sections) >= 1
assert sections[0].name == "HPI"
assert "cough" in sections[0].content
def test_detects_multiple_sections(self):
note = "HPI: Patient has fever.\nPhysical Examination: Lungs clear.\nChest X-Ray: Normal."
sections = self.detector.detect(note)
assert len(sections) == 3
assert sections[0].name == "HPI"
assert sections[1].name == "PHYSICAL_EXAM"
assert sections[2].name == "IMAGING"
def test_handles_no_sections(self):
note = "This is just plain text without section headers."
sections = self.detector.detect(note)
assert len(sections) == 0
def test_handles_case_insensitive_headers(self):
note = "PHYSICAL EXAMINATION:\nWheezing noted."
sections = self.detector.detect(note)
assert len(sections) == 1
assert sections[0].name == "PHYSICAL_EXAM"
class TestFindingExtractor:
def setup_method(self):
self.extractor = FindingExtractor()
def test_extracts_wheezing(self):
sections = [SectionResult(name="PHYSICAL_EXAM", content="Bilateral wheezing noted.", start_pos=0, end_pos=25)]
matches = self.extractor.extract(sections)
assert "wheezing" in [m.name for m in matches]
def test_extracts_crackles(self):
sections = [SectionResult(name="PHYSICAL_EXAM", content="Crackles in right lower lobe.", start_pos=0, end_pos=30)]
matches = self.extractor.extract(sections)
assert "crackles" in [m.name for m in matches]
def test_extracts_consolidation_from_imaging(self):
sections = [SectionResult(name="IMAGING", content="Right lower lobe consolidation.", start_pos=0, end_pos=32)]
matches = self.extractor.extract(sections)
assert "consolidation" in [m.name for m in matches]
def test_no_false_positives_on_clean_text(self):
sections = [SectionResult(name="HPI", content="Patient feels well today.", start_pos=0, end_pos=25)]
matches = self.extractor.extract(sections)
assert len(matches) == 0
class TestNegationDetector:
def setup_method(self):
self.detector = NegationDetector()
def test_detects_no_wheezing(self):
match = RawFindingMatch(name="wheezing", text_span="wheezing", start_pos=3, end_pos=11, section_name="PHYSICAL_EXAM")
findings = self.detector.apply([match], "No wheezing noted.")
assert findings[0].is_negated is True
def test_affirmed_finding(self):
match = RawFindingMatch(name="wheezing", text_span="wheezing", start_pos=11, end_pos=19, section_name="PHYSICAL_EXAM")
findings = self.detector.apply([match], "Bilateral wheezing noted.")
assert findings[0].is_negated is False
def test_denies_negation(self):
match = RawFindingMatch(name="fever", text_span="fever", start_pos=8, end_pos=13, section_name="HPI")
findings = self.detector.apply([match], "Denies fever or chills.")
assert findings[0].is_negated is True
class TestPipelineIntegration:
def setup_method(self):
self.pipeline = ClinicalPipeline()
def test_full_pipeline_pneumonia_note(self):
note = "History of Present Illness:\nPatient presents with fever and cough.\nPhysical Examination:\nCrackles in right lower lobe. No wheezing.\nChest X-Ray:\nRight lower lobe consolidation."
output = self.pipeline.process(note)
assert len(output.errors) == 0
assert len(output.sections) == 3
finding_names = [f.name for f in output.findings]
assert "crackles" in finding_names
assert "wheezing" in finding_names
assert "consolidation" in finding_names
wheezing = [f for f in output.findings if f.name == "wheezing"][0]
assert wheezing.is_negated is True
crackles = [f for f in output.findings if f.name == "crackles"][0]
assert crackles.is_negated is False
def test_pipeline_returns_evidence_dict(self):
note = "Physical Examination:\nBilateral wheezing. No crackles."
output = self.pipeline.process(note)
evidence = output.to_evidence_dict()
assert evidence["Wheezing"] == "present"
assert evidence["Crackles"] == "absent"Your pipeline is now battle-tested. Next up, you will compute Conditional Probability Tables from extracted findings to power the Bayesian network.
Compute CPTs and Build the Bayesian Network
Your NLP pipeline can now extract structured findings from raw clinical notes. The next question is: what do those findings actually tell us about the patient's diagnosis?
In this step, you will bridge the gap between extracted evidence and probabilistic reasoning. You will compute Conditional Probability Tables (CPTs) from your synthetic data, then use those tables to build a Bayesian network that can answer diagnostic queries like "given wheezing and dyspnea, what is the probability of asthma?"
In this step, get ready to:
- Compute conditional probability tables from pipeline-extracted findings.
- Build a Bayesian network using pgmpy with disease and finding nodes.
- Run probabilistic inference to answer clinical diagnostic queries.
Compute conditional probability tables
A CPT answers the question: "If a patient has disease X, what is the probability they show finding Y?" You will count how often each finding appears alongside each disease in your synthetic data, then normalize those counts into probabilities.
- Open cpt_builder.py in VS Code and add the compute_cpts function by pasting this code:
"""Build Conditional Probability Tables from pipeline-extracted findings."""
from collections import defaultdict
from models import PipelineOutput
def compute_cpts(
pipeline_outputs: list[PipelineOutput],
ground_truth_diseases: list[str],
finding_names: list[str],
disease_names: list[str],
) -> dict[str, dict[str, dict[str, float]]]:
"""Compute P(Finding=present|Disease) and P(Finding=absent|Disease) from data.
Returns:
Dict of {finding_name: {disease: {"present": prob, "absent": prob}}}
"""
# Count occurrences
disease_counts: dict[str, int] = defaultdict(int)
finding_given_disease: dict[tuple, dict[str, int]] = defaultdict(lambda: defaultdict(int))
for output, disease in zip(pipeline_outputs, ground_truth_diseases):
disease_counts[disease] += 1
# Get affirmed finding names for this note
affirmed = {f.name for f in output.get_affirmed_findings()}
for finding_name in finding_names:
if finding_name in affirmed:
finding_given_disease[(finding_name, disease)]["present"] += 1
else:
finding_given_disease[(finding_name, disease)]["absent"] += 1
# Convert counts to probabilities with Laplace smoothing
cpts: dict[str, dict[str, dict[str, float]]] = {}
for finding_name in finding_names:
cpts[finding_name] = {}
for disease in disease_names:
total = disease_counts[disease] + 2 # Laplace smoothing
present_count = finding_given_disease[(finding_name, disease)].get("present", 0) + 1
absent_count = finding_given_disease[(finding_name, disease)].get("absent", 0) + 1
cpts[finding_name][disease] = {
"present": present_count / total,
"absent": absent_count / total,
}
return cptsWhat does this code do?
The function loops through every pipeline output and its ground-truth disease label. For each note, it collects the affirmed findings (those not negated) and counts how many times each finding appears with each disease.
It then converts raw counts into probabilities using Laplace smoothing (adding 1 to every count). Laplace smoothing prevents zero probabilities, which would cause the Bayesian network to produce degenerate results from a small dataset.
You won't see a change yet. The next chunk adds a companion function that computes disease priors.
- Below the compute_cpts function in cpt_builder.py, add the compute_disease_prior function:
def compute_disease_prior(
ground_truth_diseases: list[str],
disease_names: list[str],
) -> dict[str, float]:
"""Compute P(Disease) prior from ground truth labels."""
total = len(ground_truth_diseases) + len(disease_names) # Laplace
counts: dict[str, int] = defaultdict(lambda: 1) # Start at 1 for Laplace smoothing
for d in ground_truth_diseases:
counts[d] += 1
return {d: counts[d] / total for d in disease_names}What does this code do?
This computes P(Disease) from how often each disease appears in the ground-truth labels. It uses the same Laplace smoothing approach, starting every count at 1 so no disease has zero prior probability.
- Save cpt_builder.py.
- Verify the module loads correctly by running this in your terminal:
python -c "from cpt_builder import compute_cpts, compute_disease_prior; print('cpt_builder imports OK')"You should see cpt_builder imports OK printed to the terminal.
Seeing an ImportError?
Make sure you are running the command from inside your respiratory_pipeline folder where models.py lives. The cpt_builder.py file imports from models, so Python needs to find that module in the same directory.
help me debug a cpt_builder ImportError
✔️ Awesome, I've got everything!
Great. Double check you have saved your file before moving on.
ⓧ I'd like to double check the full code
"""Build Conditional Probability Tables from pipeline-extracted findings."""
from collections import defaultdict
from models import PipelineOutput
def compute_cpts(
pipeline_outputs: list[PipelineOutput],
ground_truth_diseases: list[str],
finding_names: list[str],
disease_names: list[str],
) -> dict[str, dict[str, dict[str, float]]]:
"""Compute P(Finding=present|Disease) and P(Finding=absent|Disease) from data.
Returns:
Dict of {finding_name: {disease: {"present": prob, "absent": prob}}}
"""
# Count occurrences
disease_counts: dict[str, int] = defaultdict(int)
finding_given_disease: dict[tuple, dict[str, int]] = defaultdict(lambda: defaultdict(int))
for output, disease in zip(pipeline_outputs, ground_truth_diseases):
disease_counts[disease] += 1
# Get affirmed finding names for this note
affirmed = {f.name for f in output.get_affirmed_findings()}
for finding_name in finding_names:
if finding_name in affirmed:
finding_given_disease[(finding_name, disease)]["present"] += 1
else:
finding_given_disease[(finding_name, disease)]["absent"] += 1
# Convert counts to probabilities with Laplace smoothing
cpts: dict[str, dict[str, dict[str, float]]] = {}
for finding_name in finding_names:
cpts[finding_name] = {}
for disease in disease_names:
total = disease_counts[disease] + 2 # Laplace smoothing
present_count = finding_given_disease[(finding_name, disease)].get("present", 0) + 1
absent_count = finding_given_disease[(finding_name, disease)].get("absent", 0) + 1
cpts[finding_name][disease] = {
"present": present_count / total,
"absent": absent_count / total,
}
return cpts
def compute_disease_prior(
ground_truth_diseases: list[str],
disease_names: list[str],
) -> dict[str, float]:
"""Compute P(Disease) prior from ground truth labels."""
total = len(ground_truth_diseases) + len(disease_names) # Laplace
counts: dict[str, int] = defaultdict(lambda: 1) # Start at 1 for Laplace smoothing
for d in ground_truth_diseases:
counts[d] += 1
return {d: counts[d] / total for d in disease_names}Build the Bayesian network structure
Now you have a way to compute probabilities from data. The next piece is a pgmpy Bayesian network that encodes those probabilities as a directed graph. The structure uses a naive Bayes layout: a single Disease node with four states (Pneumonia, COPD Exacerbation, Asthma Attack, Heart Failure) acts as a parent for each binary Finding node.
- Open bayesian_network.py in VS Code and paste the following code to define the network constants and the build function:
"""Respiratory disease Bayesian network using pgmpy."""
from pgmpy.models import DiscreteBayesianNetwork
from pgmpy.factors.discrete import TabularCPD
from pgmpy.inference import VariableElimination
# Disease and finding node names
DISEASES = ["Pneumonia", "COPD_Exacerbation", "Asthma_Attack", "Heart_Failure"]
FINDINGS = ["Wheezing", "Crackles", "Consolidation", "Effusion", "Fever", "Dyspnea"]
def build_respiratory_network(
disease_priors: dict[str, float],
cpts: dict[str, dict[str, dict[str, float]]],
) -> DiscreteBayesianNetwork:
"""Build the Bayesian network from computed CPTs.
Structure: Each Disease node is a parent of each Finding node.
This is a naive Bayes structure (findings are conditionally
independent given the disease).
For simplicity, we use a single Disease node with 4 states
rather than 4 separate binary disease nodes.
"""
# Define network structure: Disease -> each Finding
edges = [("Disease", finding) for finding in FINDINGS]
model = DiscreteBayesianNetwork(edges)
# Disease prior CPD
disease_values = [[disease_priors[d]] for d in DISEASES]
cpd_disease = TabularCPD(
variable="Disease",
variable_card=len(DISEASES),
values=disease_values,
state_names={"Disease": DISEASES},
)
# Finding CPDs: P(Finding | Disease)
finding_cpds = []
for finding in FINDINGS:
# Map finding name to CPT key (lowercase with underscores)
cpt_key = finding.lower()
# Build values matrix: rows = finding states, cols = disease states
present_probs = []
absent_probs = []
for disease in DISEASES:
disease_key = disease.lower().replace("_", " ")
# Look up CPT, fall back to uniform if not found
if cpt_key in cpts and disease_key in cpts[cpt_key]:
present_probs.append(cpts[cpt_key][disease_key]["present"])
absent_probs.append(cpts[cpt_key][disease_key]["absent"])
else:
present_probs.append(0.5)
absent_probs.append(0.5)
cpd_finding = TabularCPD(
variable=finding,
variable_card=2,
values=[absent_probs, present_probs],
evidence=["Disease"],
evidence_card=[len(DISEASES)],
state_names={
finding: ["absent", "present"],
"Disease": DISEASES,
},
)
finding_cpds.append(cpd_finding)
# Assemble model
model.add_cpds(cpd_disease, *finding_cpds)
# Validate
assert model.check_model(), "Model validation failed!"
return modelWhat does this code do?
- DISEASES and FINDINGS define the node names in the network. The Disease node has 4 states. Each Finding node is binary (present or absent).
- DiscreteBayesianNetwork(edges) creates the directed graph with edges from Disease to each Finding.
- TabularCPD objects encode the probability tables. The Disease CPD holds the prior probabilities. Each Finding CPD holds P(Finding | Disease) values from your computed CPTs.
- check_model() validates that all CPDs are consistent: correct parent sets, probability columns sum to 1, and cardinalities match.
Why naive Bayes instead of a richer structure?
A naive Bayes structure assumes findings are conditionally independent given the disease. In reality, crackles and consolidation often co-occur in pneumonia. A richer model would add edges between findings, but that requires exponentially more training data to estimate the CPTs reliably.
With only 5 synthetic notes, the naive Bayes assumption keeps the model tractable and the CPTs well-estimated. The focus of this project is the full pipeline integration, not Bayesian network complexity.
You won't see a change yet. The next chunk adds the inference function that queries the network.
Add inference and validate the full network
The build function creates the network, but you still need a way to ask it questions. The run_inference function takes observed findings as evidence and returns the posterior probability of each disease using Variable Elimination.
- Below the build_respiratory_network function in bayesian_network.py, add the inference function:
def run_inference(
model: DiscreteBayesianNetwork,
evidence: dict[str, str],
) -> dict[str, float]:
"""Run Variable Elimination inference on the model.
Args:
model: The fitted Bayesian network.
evidence: Dict of {FindingNode: "present"|"absent"}.
Returns:
Dict of {disease_name: probability}.
"""
infer = VariableElimination(model)
# Filter evidence to only include nodes that exist in the model
valid_evidence = {
k: v for k, v in evidence.items() if k in model.nodes()
}
if not valid_evidence:
# No valid evidence, return priors
result = infer.query(variables=["Disease"])
else:
result = infer.query(variables=["Disease"], evidence=valid_evidence)
# Extract probabilities
probs = {}
for i, disease in enumerate(DISEASES):
probs[disease] = float(result.values[i])
return probsWhat does this code do?
- VariableElimination(model) creates an inference engine that can compute exact posterior probabilities by marginalizing over unobserved nodes.
- The function filters evidence to only include nodes the model knows about, preventing errors from unmapped findings.
- infer.query(variables=["Disease"], evidence=valid_evidence) computes P(Disease | observed findings) and returns a DiscreteFactor whose .values attribute holds the probability array.
- Save bayesian_network.py.
- Verify the full pipeline-to-network flow works by running this command:
python -c "
from pipeline import ClinicalPipeline
from cpt_builder import compute_cpts, compute_disease_prior
from bayesian_network import build_respiratory_network, run_inference, DISEASES, FINDINGS
from synthetic_data import get_training_notes
pipeline = ClinicalPipeline()
data = get_training_notes()
outputs = [pipeline.process(e['note']) for e in data]
diseases = [e['disease'] for e in data]
finding_names = [f.lower() for f in FINDINGS]
disease_names = [d.lower().replace('_', ' ') for d in DISEASES]
cpts = compute_cpts(outputs, diseases, finding_names, disease_names)
priors = compute_disease_prior(diseases, disease_names)
disease_priors = {d: priors.get(d.lower().replace('_', ' '), 0.25) for d in DISEASES}
model = build_respiratory_network(disease_priors, cpts)
result = run_inference(model, {'Wheezing': 'present', 'Crackles': 'absent'})
print('Model validated and inference works!')
for disease, prob in result.items():
print(f' {disease}: {prob:.3f}')
"You should see Model validated and inference works! followed by four disease probabilities. Asthma_Attack and COPD_Exacerbation should have the highest values since wheezing is characteristic of both.
Seeing a model validation error?
If you see AssertionError: Model validation failed!, the CPD values likely do not sum to 1 per column. Check that your compute_cpts function uses disease_counts[disease] + 2 as the denominator (2 for Laplace smoothing across present and absent).
If you see an ImportError mentioning BayesianNetwork, make sure your import uses DiscreteBayesianNetwork from pgmpy.models. The older BayesianNetwork class was removed in pgmpy 1.0.0.
help me debug my Bayesian network build error
✔️ Awesome, I've got everything!
Great. Double check you have saved your file before moving on.
ⓧ I'd like to double check the full code
"""Respiratory disease Bayesian network using pgmpy."""
from pgmpy.models import DiscreteBayesianNetwork
from pgmpy.factors.discrete import TabularCPD
from pgmpy.inference import VariableElimination
# Disease and finding node names
DISEASES = ["Pneumonia", "COPD_Exacerbation", "Asthma_Attack", "Heart_Failure"]
FINDINGS = ["Wheezing", "Crackles", "Consolidation", "Effusion", "Fever", "Dyspnea"]
def build_respiratory_network(
disease_priors: dict[str, float],
cpts: dict[str, dict[str, dict[str, float]]],
) -> DiscreteBayesianNetwork:
"""Build the Bayesian network from computed CPTs.
Structure: Each Disease node is a parent of each Finding node.
This is a naive Bayes structure (findings are conditionally
independent given the disease).
For simplicity, we use a single Disease node with 4 states
rather than 4 separate binary disease nodes.
"""
# Define network structure: Disease -> each Finding
edges = [("Disease", finding) for finding in FINDINGS]
model = DiscreteBayesianNetwork(edges)
# Disease prior CPD
disease_values = [[disease_priors[d]] for d in DISEASES]
cpd_disease = TabularCPD(
variable="Disease",
variable_card=len(DISEASES),
values=disease_values,
state_names={"Disease": DISEASES},
)
# Finding CPDs: P(Finding | Disease)
finding_cpds = []
for finding in FINDINGS:
# Map finding name to CPT key (lowercase with underscores)
cpt_key = finding.lower()
# Build values matrix: rows = finding states, cols = disease states
present_probs = []
absent_probs = []
for disease in DISEASES:
disease_key = disease.lower().replace("_", " ")
# Look up CPT, fall back to uniform if not found
if cpt_key in cpts and disease_key in cpts[cpt_key]:
present_probs.append(cpts[cpt_key][disease_key]["present"])
absent_probs.append(cpts[cpt_key][disease_key]["absent"])
else:
present_probs.append(0.5)
absent_probs.append(0.5)
cpd_finding = TabularCPD(
variable=finding,
variable_card=2,
values=[absent_probs, present_probs],
evidence=["Disease"],
evidence_card=[len(DISEASES)],
state_names={
finding: ["absent", "present"],
"Disease": DISEASES,
},
)
finding_cpds.append(cpd_finding)
# Assemble model
model.add_cpds(cpd_disease, *finding_cpds)
# Validate
assert model.check_model(), "Model validation failed!"
return model
def run_inference(
model: DiscreteBayesianNetwork,
evidence: dict[str, str],
) -> dict[str, float]:
"""Run Variable Elimination inference on the model.
Args:
model: The fitted Bayesian network.
evidence: Dict of {FindingNode: "present"|"absent"}.
Returns:
Dict of {disease_name: probability}.
"""
infer = VariableElimination(model)
# Filter evidence to only include nodes that exist in the model
valid_evidence = {
k: v for k, v in evidence.items() if k in model.nodes()
}
if not valid_evidence:
# No valid evidence, return priors
result = infer.query(variables=["Disease"])
else:
result = infer.query(variables=["Disease"], evidence=valid_evidence)
# Extract probabilities
probs = {}
for i, disease in enumerate(DISEASES):
probs[disease] = float(result.values[i])
return probsYour probabilistic reasoning layer is now complete. The Bayesian network can take extracted findings and return ranked disease probabilities. Next up, you will wire the entire pipeline end-to-end so a single function call takes a raw clinical note and returns a diagnosis.
Integrate Everything into the diagnose() Function
You now have every building block of the pipeline working independently. The NLP stages extract findings from clinical notes, the CPT builder computes probabilities from data, and the Bayesian network runs inference queries.
In this final step, you will wire everything together into a single diagnose() function that takes a raw clinical note and returns ranked disease probabilities. This is the integration point where NLP meets probabilistic reasoning.
In this step, get ready to:
- Wire all pipeline modules into a training function that builds the Bayesian network from synthetic data.
- Create a diagnose() function that runs a note through NLP extraction and into Bayesian inference.
- Run the full pipeline on 3 clinical scenarios and verify the output makes clinical sense.
Wire all pipeline modules into a training function
The training function is the glue between your NLP pipeline and your Bayesian network. It processes all synthetic notes, extracts findings, computes CPTs from those findings, and builds the network.
- Open main.py in VS Code and add the following imports and train_model() function:
"""End-to-end pipeline integration: from clinical note to diagnosis."""
from pipeline import ClinicalPipeline
from cpt_builder import compute_cpts, compute_disease_prior
from bayesian_network import build_respiratory_network, run_inference, DISEASES, FINDINGS
from synthetic_data import get_training_notes
def train_model():
# Create a pipeline instance and load synthetic training data
pipeline = ClinicalPipeline()
training_data = get_training_notes()
# Process each note through the NLP pipeline
outputs = []
diseases = []
for entry in training_data:
output = pipeline.process(entry["note"])
outputs.append(output)
diseases.append(entry["disease"])
# Map node names to lowercase for CPT computation
finding_names = [f.lower() for f in FINDINGS]
disease_names = [d.lower().replace("_", " ") for d in DISEASES]
# Compute CPTs and priors from extracted findings
cpts = compute_cpts(outputs, diseases, finding_names, disease_names)
priors = compute_disease_prior(diseases, disease_names)
# Map priors back to BN disease node names
disease_priors = {}
for d in DISEASES:
key = d.lower().replace("_", " ")
disease_priors[d] = priors.get(key, 1.0 / len(DISEASES))
# Build the Bayesian network with computed values
model = build_respiratory_network(disease_priors, cpts)
return model, pipelineWhat does this code do?
The imports pull in every module you have built so far. This is where all your previous work converges into a single entry point.
The train_model() function does four things: it runs all 5 synthetic notes through your NLP pipeline, computes CPTs from the extracted findings, computes disease priors from the ground-truth labels, and builds the DiscreteBayesianNetwork. It returns both the model and the pipeline so they can be reused for inference.
- Save main.py.
- You won't see output yet. The next chunk adds the diagnose() function that uses this trained model.
Seeing an import error?
Make sure all your module files (pipeline.py, cpt_builder.py, bayesian_network.py, synthetic_data.py) are in the same folder as main.py. Python resolves imports relative to the file's directory.
Help me debug an import error in main.py
Create the diagnose() function
This is the single function a user would call to get a diagnosis from a clinical note. It orchestrates the full flow: NLP extraction, evidence conversion, and Bayesian inference.
- Below the train_model() function, add the diagnose() function:
def diagnose(note: str, model, pipeline: ClinicalPipeline) -> dict[str, float]:
# Run the NLP pipeline on the raw note
output = pipeline.process(note)
# Convert extracted findings to BN evidence format
evidence = output.to_evidence_dict()
# Filter to only findings the Bayesian network knows about
valid_evidence = {k: v for k, v in evidence.items() if k in FINDINGS}
# Run probabilistic inference
probs = run_inference(model, valid_evidence)
# Sort by probability, highest first
ranked = dict(sorted(probs.items(), key=lambda x: x[1], reverse=True))
return rankedWhat does this code do?
- pipeline.process(note) runs the full NLP chain: section detection, finding extraction, and negation handling.
- to_evidence_dict() converts findings into the format the Bayesian network expects (e.g., {"Wheezing": "present", "Crackles": "absent"}).
- The filter against FINDINGS drops any extracted findings (like cough or rhonchi) that the network has no node for.
- The result is sorted so the most likely disease appears first.
- Save main.py.
- The next chunk adds the main() function that runs real scenarios through diagnose().
Unsure about the evidence filtering?
Your NLP pipeline extracts 9 different findings (wheezing, crackles, rhonchi, consolidation, effusion, decreased_breath_sounds, fever, dyspnea, cough). But your Bayesian network only has nodes for 6 of them (Wheezing, Crackles, Consolidation, Effusion, Fever, Dyspnea). The filter ensures you only pass evidence the network can use.
Help me understand why diagnose() filters evidence against FINDINGS
Run the full pipeline on 3 clinical scenarios
Now you will add a main() function that trains the model and runs diagnostic inference on three different clinical scenarios: pneumonia, asthma, and heart failure.
- Below the diagnose() function, add the main() function and entry point:
def main():
"""Demonstrate the full end-to-end pipeline."""
print("=" * 60)
print("RESPIRATORY DISEASE NLP + BAYESIAN NETWORK PIPELINE")
print("=" * 60)
# Step 1: Train the model from synthetic data
print("\n[1] Training model from synthetic clinical notes...")
model, pipeline = train_model()
print(" Model built successfully!")
print(f" Nodes: {list(model.nodes())}")
print(f" Edges: {list(model.edges())}")
# Step 2: Define test scenarios with different clinical presentations
test_notes = [
("Pneumonia Scenario", "History of Present Illness:\n60-year-old with fever and productive cough for 4 days.\nPhysical Examination:\nCrackles at right base. No wheezing. Decreased breath sounds.\nChest X-Ray:\nRight lower lobe consolidation. Small effusion."),
("Asthma Scenario", "HPI:\n30-year-old with acute wheezing and dyspnea. Denies fever.\nPE:\nDiffuse wheezing bilaterally. No crackles.\nCXR:\nClear. No consolidation. No effusion."),
("Heart Failure Scenario", "History of Present Illness:\n75-year-old with progressive dyspnea and orthopnea. No fever.\nPhysical Examination:\nBilateral crackles. No wheezing. JVD present.\nChest X-Ray:\nBilateral effusions. No consolidation."),
]
# Step 3: Run inference on each scenario
print("\n[2] Running diagnostic inference on test cases...\n")
for scenario_name, note in test_notes:
print(f"--- {scenario_name} ---")
results = diagnose(note, model, pipeline)
for disease, prob in results.items():
bar = "X" * int(prob * 30)
print(f" {disease:<22} {prob:.3f} {bar}")
print()
print("=" * 60)
print("Pipeline complete!")
print("=" * 60)
if __name__ == "__main__":
main()What does this code do?
- train_model() processes all 5 synthetic notes, computes CPTs, and builds the Bayesian network.
- Each test scenario is a realistic clinical note designed to trigger a different top-ranked disease.
- The output prints each disease with its posterior probability and a simple bar chart so you can visually compare likelihoods.
- Save main.py.
- Run the full pipeline by running this command:
python main.py- You should see output for all 3 scenarios. The Pneumonia Scenario should rank Pneumonia highest, the Asthma Scenario should rank Asthma_Attack highest, and the Heart Failure Scenario should rank Heart_Failure highest.
What should I see?
The output shows a header, model structure info (7 nodes, 6 edges), then three scenario blocks. Each scenario lists 4 diseases with their posterior probabilities (0.000 to 1.000) and a bar made of X characters. The probabilities for each scenario sum to 1.0.
Exact probability values depend on how the Laplace-smoothed CPTs shake out from your 5 training notes. The key check is that the top disease matches the scenario.
Top disease doesn't match the scenario?
Check that your synthetic_data.py contains all 5 notes with the correct disease labels. The CPTs are computed from this data, so a missing or mislabeled note shifts the probabilities.
Also confirm your negation_detector.py is correctly marking negated findings. If "No wheezing" is treated as affirmed, it corrupts the evidence passed to the network.
Help me debug why my pipeline's top prediction doesn't match the expected disease
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py.
ⓧ I'd like to double check the full code
"""End-to-end pipeline integration: from clinical note to diagnosis."""
from pipeline import ClinicalPipeline
from cpt_builder import compute_cpts, compute_disease_prior
from bayesian_network import build_respiratory_network, run_inference, DISEASES, FINDINGS
from synthetic_data import get_training_notes
def train_model():
pipeline = ClinicalPipeline()
training_data = get_training_notes()
outputs = []
diseases = []
for entry in training_data:
output = pipeline.process(entry["note"])
outputs.append(output)
diseases.append(entry["disease"])
finding_names = [f.lower() for f in FINDINGS]
disease_names = [d.lower().replace("_", " ") for d in DISEASES]
cpts = compute_cpts(outputs, diseases, finding_names, disease_names)
priors = compute_disease_prior(diseases, disease_names)
disease_priors = {}
for d in DISEASES:
key = d.lower().replace("_", " ")
disease_priors[d] = priors.get(key, 1.0 / len(DISEASES))
model = build_respiratory_network(disease_priors, cpts)
return model, pipeline
def diagnose(note: str, model, pipeline: ClinicalPipeline) -> dict[str, float]:
output = pipeline.process(note)
evidence = output.to_evidence_dict()
valid_evidence = {k: v for k, v in evidence.items() if k in FINDINGS}
probs = run_inference(model, valid_evidence)
ranked = dict(sorted(probs.items(), key=lambda x: x[1], reverse=True))
return ranked
def main():
print("=" * 60)
print("RESPIRATORY DISEASE NLP + BAYESIAN NETWORK PIPELINE")
print("=" * 60)
print("\n[1] Training model from synthetic clinical notes...")
model, pipeline = train_model()
print(" Model built successfully!")
print(f" Nodes: {list(model.nodes())}")
print(f" Edges: {list(model.edges())}")
test_notes = [
("Pneumonia Scenario", "History of Present Illness:\n60-year-old with fever and productive cough for 4 days.\nPhysical Examination:\nCrackles at right base. No wheezing. Decreased breath sounds.\nChest X-Ray:\nRight lower lobe consolidation. Small effusion."),
("Asthma Scenario", "HPI:\n30-year-old with acute wheezing and dyspnea. Denies fever.\nPE:\nDiffuse wheezing bilaterally. No crackles.\nCXR:\nClear. No consolidation. No effusion."),
("Heart Failure Scenario", "History of Present Illness:\n75-year-old with progressive dyspnea and orthopnea. No fever.\nPhysical Examination:\nBilateral crackles. No wheezing. JVD present.\nChest X-Ray:\nBilateral effusions. No consolidation."),
]
print("\n[2] Running diagnostic inference on test cases...\n")
for scenario_name, note in test_notes:
print(f"--- {scenario_name} ---")
results = diagnose(note, model, pipeline)
for disease, prob in results.items():
bar = "X" * int(prob * 30)
print(f" {disease:<22} {prob:.3f} {bar}")
print()
print("=" * 60)
print("Pipeline complete!")
print("=" * 60)
if __name__ == "__main__":
main()Your pipeline is complete. You have a working system that takes a raw clinical note, extracts structured findings using NLP, and feeds them into a Bayesian network to compute disease probabilities. Next up is the Secret Mission, where you will calibrate this model against known outcomes.
Secret mission
Confidence Calibration
Your Bayesian network produces probabilities, but how accurate are those probabilities? A model that says 80% confidence should be correct 80% of the time. Build a calibration module that measures how well your pipeline's predictions match reality.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your project files and installed packages, or delete them entirely.
Resources you used:
- The respiratory_pipeline project folder on your Desktop (all Python source files, test files, and synthetic data).
- pip packages: pgmpy and pytest.
Keep everything running
No action needed. Choose this if you plan to extend the pipeline with more findings, additional diseases, or real clinical data.
- Your project folder stays on your Desktop.
- Your pip packages remain installed and available for other Python projects.
Pause - I'll come back to this later
Since this project is entirely local with no running processes, there is nothing to pause. Your files and packages stay where they are until you return.
- Close VS Code to free up memory.
- Your project folder and pip packages remain intact for when you come back.
Delete - I don't want to use this again
Remove all project files and uninstall the packages you added during this project.
- Delete the project folder by running this command in your terminal:
macOS
rm -rf ~/Desktop/respiratory_pipelineWindows
Remove-Item -Recurse -Force ~\Desktop\respiratory_pipeline- Uninstall the pip packages by running:
pip uninstall pgmpy pytest -yNice Work!
Nice Work!
You just built a healthcare AI system from scratch. Starting from raw clinical text, you engineered a complete pipeline that processes physician notes through multiple NLP stages and feeds structured evidence into a probabilistic reasoning engine for respiratory disease diagnosis.
You've learned how to:
- Build a clinical NLP pipeline that chains section detection, finding extraction, and negation handling to transform unstructured physician notes into structured evidence.
- Compute Conditional Probability Tables from extracted findings and assemble a Bayesian network with pgmpy for respiratory disease diagnosis.
- Integrate all components into a single diagnose() function that takes a raw clinical note and returns ranked disease probabilities using Variable Elimination inference.
- Calibrated your model against ground truth diagnoses and identified where the network is overconfident or underconfident.
Ready to quiz yourself?