Lab L3.1: Architecture Design

🎯 Assignment: Accept this lab on GitHub Classroom
You’ll get your own repository with starter code, instructions, and automatic grading.

Duration	120 minutes
Prerequisites	Previous module completed

Objectives

• Design scalable agent architecture
• Implement gateway pattern
• Handle complex routing

---

How to Complete This Lab

1. Accept the Assignment — Click the GitHub Classroom link above
2. Clone Your Repo — git clone <your-repo-url>
3. Read the README — Your repo has detailed requirements and grading criteria
4. Write Your Code — Implement the solution in solution/agent.py
5. Test Locally — Use swaig-test to verify your agent works
6. Push to Submit — git push triggers auto-grading

---

Key Concepts

The following exercises walk through the concepts you’ll need. Your GitHub Classroom repo README has the specific requirements for grading.

Scenario

You’re architecting a voice AI system for “HealthFirst Insurance” that needs to handle:

• Member services (benefits, claims, ID cards)
• Provider services (eligibility, prior auth)
• General inquiries (hours, locations, contact info)

---

Part 1: Architecture Design (45 min)

Task

Design the system architecture addressing:

1. Agent organization (how many, what roles)
2. Routing strategy
3. State management
4. Security requirements (HIPAA)

Deliverable

Create architecture_design.md with:

# HealthFirst Voice AI Architecture

## Overview
[High-level description]

## System Diagram
[ASCII or description of component relationships]

## Agents

### Gateway Agent

- Route: /
- Purpose: [describe]
- Functions: [list]

### Member Services Agent

- Route: /member
- Purpose: [describe]
- Functions: [list]

### Provider Services Agent

- Route: /provider
- Purpose: [describe]
- Functions: [list]

## Data Flow
[How calls move through the system]

## State Management
[How state is shared/persisted]

## Security Considerations
[HIPAA compliance approach]

Design Questions to Answer

1. Should member and provider services be separate agents?
2. How do you verify caller identity for PHI access?
3. Where does session state live?
4. How do you handle after-hours calls?

---

Part 2: Architecture Decision Records (30 min)

Task

Document three key decisions using ADR format.

Template

# ADR [NUMBER]: [TITLE]

## Status
[Proposed | Accepted | Deprecated | Superseded]

## Context
[What is the issue we're addressing?]

## Decision
[What did we decide to do?]

## Consequences
[What becomes easier or harder?]

Required ADRs

ADR 001: Agent Separation Strategy

• Decide: One agent vs. multiple specialized agents
• Consider: Maintenance, security boundaries, prompt complexity

ADR 002: Caller Verification Approach

• Decide: How to verify identity for PHI access
• Consider: PIN, callback, multi-factor

ADR 003: State Persistence

• Decide: In-memory vs. Redis vs. Database
• Consider: Scalability, cost, complexity

Example ADR

# ADR 001: Multi-Agent Architecture

## Status
Accepted

## Context
HealthFirst needs to handle member services, provider services,
and general inquiries. We need to decide how to organize the agents.

## Decision
We will use a multi-agent architecture with:
- Gateway agent for routing and general inquiries
- Member services agent for member-specific functions
- Provider services agent for healthcare provider functions

## Consequences
### Positive

- Clear separation of concerns
- Teams can work independently on each agent
- Security boundaries between member and provider data
- Easier prompt management (focused contexts)

### Negative

- More complex deployment (3 agents vs 1)
- Need shared state solution for cross-agent data
- More endpoints to monitor

---

Part 3: Anti-Pattern Identification (25 min)

Task

Review the following code and identify anti-patterns. Document each issue and propose a fix.

Code to Review

# healthfirst_agent.py (CONTAINS ANTI-PATTERNS)

from signalwire_agents import AgentBase, SwaigFunctionResult
import requests

class HealthFirstAgent(AgentBase):
    def __init__(self):
        super().__init__(name="healthfirst")

        # REVIEW: Is this a good prompt structure?
        self.prompt_add_section(
            "Role",
            "You are an extremely professional, highly knowledgeable, and "
            "exceptionally customer-focused health insurance representative "
            "who works for HealthFirst Insurance Company, a leading provider "
            "of comprehensive health coverage solutions for individuals, "
            "families, and businesses across the nation since 1985..."
            # (continues for 500 more words)
        )

        # REVIEW: Hardcoded credentials?
        self.set_params({
            "swml_basic_auth_user": "healthfirst",
            "swml_basic_auth_password": "insurance123"
        })

        self.add_language("English", "en-US", "rime.spore")

    @AgentBase.tool(description="Get member benefits")
    def get_benefits(self, args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        member_id = args.get("member_id", "")
        # REVIEW: Error handling?
        response = requests.get(
            f"https://api.healthfirst.com/members/{member_id}/benefits"
        )
        return SwaigFunctionResult(str(response.json()))

    @AgentBase.tool(description="Look up claim")
    def get_claim(self, args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        claim_id = args.get("claim_id", "")
        # REVIEW: What if this takes 30 seconds?
        response = requests.get(
            f"https://api.healthfirst.com/claims/{claim_id}"
        )
        data = response.json()
        # REVIEW: Is this appropriate error handling?
        if "error" in data:
            return SwaigFunctionResult(f"Error: {data['error']}")
        return SwaigFunctionResult(f"Claim status: {data}")

    @AgentBase.tool(description="Submit prior auth")
    def submit_prior_auth(self, args: dict, raw_data: dict = None) -> SwaigFunctionResult:
        member_id = args.get("member_id", "")
        procedure_code = args.get("procedure_code", "")
        diagnosis_code = args.get("diagnosis_code", "")
        provider_npi = args.get("provider_npi", "")
        # REVIEW: No verification before PHI access?
        response = requests.post(
            "https://api.healthfirst.com/prior-auth",
            json={
                "member_id": member_id,
                "procedure": procedure_code,
                "diagnosis": diagnosis_code,
                "provider": provider_npi
            }
        )
        return SwaigFunctionResult(f"Prior auth submitted: {response.json()}")

    # ... 45 more functions for every possible operation

Deliverable

Create anti_patterns.md:

# Anti-Pattern Analysis

## Issue 1: [Name]
**Location:** [Line/Section]
**Problem:** [Description]
**Impact:** [What goes wrong]
**Solution:** [How to fix]

## Issue 2: ...

---

Part 4: Refactored Implementation (20 min)

Task

Implement the gateway agent with proper patterns.

Deliverable: gateway_agent.py

#!/usr/bin/env python3
"""HealthFirst Gateway Agent - Properly architected."""

import os
from signalwire_agents import AgentBase, SwaigFunctionResult


class GatewayAgent(AgentBase):
    DEPARTMENTS = {
        "member": {
            "route": "/member",
            "description": "Member services - benefits, claims, ID cards"
        },
        "provider": {
            "route": "/provider",
            "description": "Provider services - eligibility, prior authorization"
        }
    }

    def __init__(self):
        super().__init__(name="healthfirst-gateway", route="/")

        self._configure_auth()
        self._configure_prompts()
        self.add_language("English", "en-US", "rime.spore")
        self._setup_functions()

    def _configure_auth(self):
        """Configure authentication from environment."""
        user = os.getenv("AUTH_USER")
        password = os.getenv("AUTH_PASSWORD")

        if not user or not password:
            raise ValueError("AUTH_USER and AUTH_PASSWORD required")

        self.set_params({
            "swml_basic_auth_user": user,
            "swml_basic_auth_password": password
        })

    def _configure_prompts(self):
        """Set up focused, concise prompts."""
        self.prompt_add_section(
            "Role",
            "HealthFirst Insurance gateway. Route callers to the right department."
        )

        self.prompt_add_section(
            "Departments",
            bullets=[
                "Member services: benefits, claims, ID cards",
                "Provider services: eligibility, prior authorization"
            ]
        )

        self.prompt_add_section(
            "Hours",
            "Member: 8am-8pm. Provider: 7am-7pm. After hours: leave callback."
        )

    def _setup_functions(self):
        @self.tool(description="List available departments")
        def list_departments(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            dept_info = [
                f"{name}: {info['description']}"
                for name, info in self.DEPARTMENTS.items()
            ]
            return SwaigFunctionResult(
                "Available departments: " + "; ".join(dept_info)
            )

        @self.tool(
            description="Route to department",
            parameters={
                "type": "object",
                "properties": {
                    "department": {
                        "type": "string",
                        "enum": list(self.DEPARTMENTS.keys())
                    }
                },
                "required": ["department"]
            }
        )
        def route_to_department(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            department = args.get("department", "")
            dept_info = self.DEPARTMENTS.get(department)
            if not dept_info:
                return SwaigFunctionResult(
                    f"Unknown department. Available: {', '.join(self.DEPARTMENTS.keys())}"
                )

            return (
                SwaigFunctionResult(f"Connecting you to {department} services.")
                .connect(dept_info["route"], final=True)
            )

        @self.tool(description="Get general information")
        def get_info(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            topic = args.get("topic", "")
            info = {
                "hours": "We're open Mon-Fri 8am-8pm for members, 7am-7pm for providers.",
                "website": "Visit healthfirst.com for online services.",
                "emergency": "For medical emergencies, hang up and call 911."
            }
            return SwaigFunctionResult(
                info.get(topic.lower(), "How can I help direct your call?")
            )


if __name__ == "__main__":
    agent = GatewayAgent()
    agent.run()

---

Validation Checklist

• Architecture diagram completed
• Three ADRs documented
• At least 5 anti-patterns identified
• Gateway agent implements proper patterns
• Environment variables used for secrets
• Prompts are concise and focused

---

Presentation (Optional)

Prepare a 5-minute presentation covering:

1. Your architecture decisions
2. Key tradeoffs
3. How you addressed HIPAA requirements
4. Lessons from anti-pattern analysis

---

Submission

Submit:

1. architecture_design.md
2. adr/ directory with ADR files
3. anti_patterns.md
4. gateway_agent.py

---

Complete Agent Code

Click to reveal complete solution

#!/usr/bin/env python3
"""Healthcare Gateway Agent - Refactored Implementation.

Lab 3.1 Deliverable: Production-ready gateway agent demonstrating
architecture patterns, security, and proper separation of concerns.

Environment variables:
    SWML_BASIC_AUTH_USER: Basic auth username (auto-detected by SDK)
    SWML_BASIC_AUTH_PASSWORD: Basic auth password (auto-detected by SDK)
"""

import os
import hashlib
import logging
from datetime import datetime
from signalwire_agents import AgentBase, SwaigFunctionResult

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class HealthcareGatewayAgent(AgentBase):
    """Gateway agent for healthcare contact center.

    Responsibilities:
    - Caller identification and verification
    - Intent classification
    - Routing to specialist agents
    """

    # Simulated patient database
    PATIENTS = {
        "+15551234567": {
            "id": "P001",
            "name": "John Smith",
            "dob": "1985-03-15",
            "ssn_last4_hash": hashlib.sha256("1234".encode()).hexdigest()
        },
        "+15559876543": {
            "id": "P002",
            "name": "Jane Doe",
            "dob": "1990-07-22",
            "ssn_last4_hash": hashlib.sha256("5678".encode()).hexdigest()
        }
    }

    MAX_VERIFICATION_ATTEMPTS = 3

    def __init__(self):
        super().__init__(name="healthcare-gateway", route="/gateway")

        self._configure_prompts()
        self._configure_global_data()
        self.add_language("English", "en-US", "rime.spore")
        self._setup_functions()

        logger.info("Healthcare Gateway Agent initialized")

    def _configure_prompts(self):
        """Configure agent prompts."""
        self.prompt_add_section(
            "Role",
            "Healthcare contact center gateway. Verify callers and route to specialists."
        )

        self.prompt_add_section(
            "Security Requirements",
            bullets=[
                "ALWAYS verify identity before discussing patient information",
                "NEVER repeat full SSN or sensitive data",
                "Pause recording for SSN collection",
                "Lock after 3 failed verification attempts"
            ]
        )

        self.prompt_add_section(
            "Departments",
            bullets=[
                "Appointments: Scheduling, rescheduling, cancellations",
                "Billing: Account balance, payments, insurance",
                "Medical: Prescription refills, general questions"
            ]
        )

    def _configure_global_data(self):
        """Set up shared configuration."""
        hour = datetime.now().hour
        self.set_global_data({
            "clinic_name": "HealthFirst Medical",
            "business_hours": "8 AM to 6 PM",
            "is_open": 8 <= hour < 18,
            "greeting": "Good morning" if hour < 12 else "Good afternoon"
        })

    def _log_security_event(self, event_type: str, data: dict):
        """Log security event without sensitive data."""
        safe_data = {k: v for k, v in data.items()
                     if k not in ['ssn', 'ssn_last4', 'dob']}
        logger.info(f"SECURITY: {event_type} - {safe_data}")

    def _verify_ssn(self, patient_id: str, ssn_last4: str) -> bool:
        """Verify SSN with timing-safe comparison."""
        for phone, patient in self.PATIENTS.items():
            if patient["id"] == patient_id:
                provided_hash = hashlib.sha256(ssn_last4.encode()).hexdigest()
                return provided_hash == patient["ssn_last4_hash"]
        return False

    def _setup_functions(self):
        """Define gateway functions."""

        @self.tool(description="Identify patient by phone number")
        def identify_by_phone(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            """Attempt to identify patient by caller ID."""
            raw_data = raw_data or {}
            caller_id = raw_data.get("caller_id_number", "")
            call_id = raw_data.get("call_id", "unknown")

            patient = self.PATIENTS.get(caller_id)

            if patient:
                self._log_security_event("PATIENT_IDENTIFIED", {
                    "call_id": call_id,
                    "patient_id": patient["id"]
                })

                global_data = self.get_global_data()
                return (
                    SwaigFunctionResult(
                        f"{global_data['greeting']}, {patient['name']}. "
                        "For security, please verify your date of birth."
                    )
                    .update_global_data({
                        "pending_patient_id": patient["id"],
                        "pending_patient_name": patient["name"],
                        "pending_dob": patient["dob"],
                        "verification_attempts": 0
                    })
                )

            return SwaigFunctionResult(
                "I don't recognize this phone number. "
                "Could you provide your patient ID or date of birth?"
            )

        @self.tool(
            description="Verify date of birth",
            parameters={
                "type": "object",
                "properties": {
                    "dob": {
                        "type": "string",
                        "description": "Date of birth (YYYY-MM-DD or spoken format)"
                    }
                },
                "required": ["dob"]
            }
        )
        def verify_dob(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            """Verify date of birth for identified patient."""
            dob = args.get("dob", "")
            raw_data = raw_data or {}
            global_data = raw_data.get("global_data", {})
            call_id = raw_data.get("call_id", "unknown")
            expected_dob = global_data.get("pending_dob")
            patient_id = global_data.get("pending_patient_id")

            if not patient_id:
                return SwaigFunctionResult(
                    "Let me first identify your account. "
                    "What is your patient ID or phone number?"
                )

            # Normalize DOB comparison
            normalized_dob = dob.replace("/", "-").strip()
            if normalized_dob == expected_dob:
                self._log_security_event("DOB_VERIFIED", {
                    "call_id": call_id,
                    "patient_id": patient_id
                })

                return (
                    SwaigFunctionResult(
                        "Thank you. For final verification, "
                        "please provide the last 4 digits of your SSN. "
                        "I'm pausing the recording for your privacy."
                    )
                    .stop_record_call(control_id="main")
                    .update_global_data({"dob_verified": True})
                )

            attempts = global_data.get("verification_attempts", 0) + 1
            if attempts >= self.MAX_VERIFICATION_ATTEMPTS:
                self._log_security_event("VERIFICATION_LOCKED", {
                    "call_id": call_id,
                    "patient_id": patient_id
                })
                return (
                    SwaigFunctionResult(
                        "Too many incorrect attempts. "
                        "Please call back or visit the clinic with ID."
                    )
                    .hangup()
                )

            return (
                SwaigFunctionResult(
                    f"That doesn't match our records. "
                    f"{self.MAX_VERIFICATION_ATTEMPTS - attempts} attempts remaining."
                )
                .update_global_data({"verification_attempts": attempts})
            )

        @self.tool(
            description="Verify last 4 digits of SSN",
            parameters={
                "type": "object",
                "properties": {
                    "ssn_last4": {
                        "type": "string",
                        "description": "Last 4 digits of SSN"
                    }
                },
                "required": ["ssn_last4"]
            },
            secure=True
        )
        def verify_ssn(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            """Verify SSN for final authentication."""
            ssn_last4 = args.get("ssn_last4", "")
            raw_data = raw_data or {}
            global_data = raw_data.get("global_data", {})
            call_id = raw_data.get("call_id", "unknown")
            patient_id = global_data.get("pending_patient_id")
            patient_name = global_data.get("pending_patient_name")

            if not global_data.get("dob_verified"):
                return SwaigFunctionResult(
                    "Please verify your date of birth first."
                )

            if self._verify_ssn(patient_id, ssn_last4):
                self._log_security_event("VERIFICATION_SUCCESS", {
                    "call_id": call_id,
                    "patient_id": patient_id
                })

                return (
                    SwaigFunctionResult(
                        f"Thank you, {patient_name}. Your identity is verified. "
                        "Recording has resumed. How can I help you today? "
                        "I can help with appointments, billing, or medical questions."
                    )
                    .record_call(control_id="main", stereo=True, format="mp3")
                    .update_global_data({
                        "verified": True,
                        "patient_id": patient_id,
                        "patient_name": patient_name,
                        "verified_at": datetime.now().isoformat()
                    })
                )

            attempts = global_data.get("verification_attempts", 0) + 1
            self._log_security_event("SSN_VERIFICATION_FAILED", {
                "call_id": call_id,
                "attempts": attempts
            })

            if attempts >= self.MAX_VERIFICATION_ATTEMPTS:
                return (
                    SwaigFunctionResult(
                        "Too many incorrect attempts. Account locked."
                    )
                    .record_call(control_id="main", stereo=True, format="mp3")
                    .hangup()
                )

            return (
                SwaigFunctionResult(
                    f"That doesn't match. {self.MAX_VERIFICATION_ATTEMPTS - attempts} attempts remaining."
                )
                .update_global_data({"verification_attempts": attempts})
            )

        @self.tool(description="Route to appointments department")
        def route_appointments(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            """Transfer to appointments specialist."""
            raw_data = raw_data or {}
            global_data = raw_data.get("global_data", {})
            if not global_data.get("verified"):
                return SwaigFunctionResult(
                    "I need to verify your identity first."
                )

            return (
                SwaigFunctionResult("Connecting you to scheduling.", post_process=True)
                .swml_transfer("/appointments", "Goodbye!", final=True)
            )

        @self.tool(description="Route to billing department")
        def route_billing(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            """Transfer to billing specialist."""
            raw_data = raw_data or {}
            global_data = raw_data.get("global_data", {})
            if not global_data.get("verified"):
                return SwaigFunctionResult(
                    "I need to verify your identity first."
                )

            return (
                SwaigFunctionResult("Connecting you to billing.", post_process=True)
                .swml_transfer("/billing", "Goodbye!", final=True)
            )

        @self.tool(description="Route to medical team")
        def route_medical(args: dict, raw_data: dict = None) -> SwaigFunctionResult:
            """Transfer to medical support."""
            raw_data = raw_data or {}
            global_data = raw_data.get("global_data", {})
            if not global_data.get("verified"):
                return SwaigFunctionResult(
                    "I need to verify your identity first."
                )

            return (
                SwaigFunctionResult("Connecting you to our medical team.", post_process=True)
                .swml_transfer("/medical", "Goodbye!", final=True)
            )


if __name__ == "__main__":
    agent = HealthcareGatewayAgent()
    agent.run()