Build a Post-Visit Medical Scribe

1
import assemblyai as aai
2
import asyncio
3
from typing import Dict, List
4
from assemblyai.types import (
5
    PIIRedactionPolicy,
6
    PIISubstitutionPolicy,
7
)
8
9
# Configure API key
10
aai.settings.api_key = "your_api_key_here"
11
12
async def transcribe_encounter_async(audio_source: str) -> Dict:
13
    """
14
    Asynchronously transcribe a medical encounter with Universal-3 Pro
15
16
    Args:
17
        audio_source: Either a local file path or publicly accessible URL
18
    """
19
    # Configure comprehensive medical transcription
20
    config = aai.TranscriptionConfig(
21
        speech_models=["universal-3-pro", "universal-2"],
22
        domain="medical-v1",
23
        language_detection=True,
24
25
        # Diarize provider and patient
26
        speaker_labels=True,
27
        speakers_expected=2,  # Typically provider and patient
28
29
        # Punctuation and Formatting
30
        punctuate=True,
31
        format_text=True,
32
33
        # Boost accuracy of medical terminology
34
        keyterms_prompt=[
35
            # Patient-specific context
36
            "hypertension", "diabetes mellitus type 2", "metformin",
37
38
            # Specialty-specific terms
39
            "auscultation", "palpation", "differential diagnosis",
40
            "chief complaint", "review of systems", "physical examination",
41
42
            # Common medications
43
            "lisinopril", "atorvastatin", "levothyroxine",
44
45
            # Procedure terms
46
            "electrocardiogram", "complete blood count", "hemoglobin A1c"
47
        ],
48
49
        # Speech understanding for medical documentation
50
        entity_detection=True,  # Extract medications, conditions, procedures
51
        redact_pii=True,  # HIPAA compliance
52
        redact_pii_policies=[
53
            PIIRedactionPolicy.person_name,
54
            PIIRedactionPolicy.date_of_birth,
55
            PIIRedactionPolicy.phone_number,
56
            PIIRedactionPolicy.email_address,
57
        ],
58
        redact_pii_sub=PIISubstitutionPolicy.hash,
59
        redact_pii_audio=True  # Create HIPAA-compliant audio
60
    )
61
62
    # Create async transcriber
63
    transcriber = aai.Transcriber()
64
65
    try:
66
        # Submit transcription job - works with both file paths and URLs
67
        transcript = await asyncio.to_thread(
68
            transcriber.transcribe,
69
            audio_source,
70
            config=config
71
        )
72
73
        # Check status
74
        if transcript.status == aai.TranscriptStatus.error:
75
            raise Exception(f"Transcription failed: {transcript.error}")
76
77
        # Process speaker-labeled utterances
78
        print("\n=== PROVIDER-PATIENT DIALOGUE ===\n")
79
80
        for utterance in transcript.utterances:
81
            # Format timestamp
82
            start_time = utterance.start / 1000  # Convert to seconds
83
            end_time = utterance.end / 1000
84
85
            # Identify speaker role
86
            speaker_label = "Provider" if utterance.speaker == "A" else "Patient"
87
88
            # Print formatted utterance
89
            print(f"[{start_time:.1f}s - {end_time:.1f}s] {speaker_label}:")
90
            print(f"  {utterance.text}")
91
            print(f"  Confidence: {utterance.confidence:.2%}\n")
92
93
        # Extract clinical entities
94
        if transcript.entities:
95
            print("\n=== CLINICAL ENTITIES DETECTED ===\n")
96
            medications = [e for e in transcript.entities if e.entity_type == "drug"]
97
            conditions = [e for e in transcript.entities if e.entity_type == "medical_condition"]
98
            procedures = [e for e in transcript.entities if e.entity_type == "medical_process"]
99
100
            if medications:
101
                print("Medications:", ", ".join([m.text for m in medications]))
102
            if conditions:
103
                print("Conditions:", ", ".join([c.text for c in conditions]))
104
            if procedures:
105
                print("Procedures:", ", ".join([p.text for p in procedures]))
106
107
        return {
108
            "transcript": transcript,
109
            "utterances": transcript.utterances,
110
            "entities": transcript.entities,
111
            "redacted_audio_url": transcript.redacted_audio_url
112
        }
113
114
    except Exception as e:
115
        print(f"Error during transcription: {e}")
116
        raise
117
118
async def main():
119
    """
120
    Example usage for medical encounter
121
    """
122
    # Can use either local file or URL
123
    audio_source = "path/to/patient_encounter.mp3"  # Or use URL
124
    # audio_source = "https://your-secure-storage.com/encounter.mp3"
125
126
    try:
127
        result = await transcribe_encounter_async(audio_source)
128
129
        # Additional processing
130
        print(f"\nEncounter duration: {result['transcript'].audio_duration} seconds")
131
132
        # Could send to LLM Gateway for SOAP note generation here
133
134
    except Exception as e:
135
        print(f"Failed to process encounter: {e}")
136
137
if __name__ == "__main__":
138
    asyncio.run(main())