src/prompt_builder.py

# src/prompt_builder.py

from typing import Any, Dict, List, Optional

__all__ = [
    "normalize_intake",
    "build_referral_summary",
]

# ------------------------------- helpers ------------------------------------


def _as_str(x: Any) -> str:
    """Coerce common containers into readable strings without adding model tokens."""
    if x is None:
        return ""
    if isinstance(x, str):
        return x.strip()
    if isinstance(x, (int, float, bool)):
        return str(x)
    if isinstance(x, dict):
        # Render simple dicts as "k: v; ..." lines
        items = [f"{k}: {_as_str(v)}" for k, v in x.items() if _as_str(v)]
        return "; ".join(items)
    if isinstance(x, (list, tuple, set)):
        items = [s for s in (_as_str(v) for v in x) if s]
        return "; ".join(items)
    return str(x)


def _first_nonempty(*candidates: Optional[str]) -> str:
    for c in candidates:
        if c and c.strip():
            return c.strip()
    return ""


def _strip_multiblank(s: str) -> str:
    # compact multiple blank lines and spaces
    out = []
    for line in s.splitlines():
        line = " ".join(line.strip().split())
        out.append(line)
    return "\n".join([l for l in out if l])


# --------------------------- normalization ----------------------------------


def normalize_intake(raw: Dict[str, Any]) -> Dict[str, Any]:
    """
    Normalize various intake shapes into a canonical dict.

    Supported shapes:
      - Nested: {'patient': {...}, 'consult': {...}}
      - Flat:   {'age', 'sex', 'chief_complaint', 'history', 'medications', 'labs', 'question', ...}

    Returns:
      {
        'patient': {...},
        'consult': {...}
      }
    """
    raw = raw or {}

    # If already nested, copy but ensure sub-dicts exist.
    if "patient" in raw or "consult" in raw:
        patient = dict(raw.get("patient") or {})
        consult = dict(raw.get("consult") or {})
    else:
        # Map flat → nested
        patient = {
            "age": raw.get("age"),
            "sex": raw.get("sex") or raw.get("gender"),
            "allergies": raw.get("allergies"),
            "medications": raw.get("medications"),
            "pmh": _first_nonempty(raw.get("pmh"), raw.get("past_medical_history")),
            "sh": _first_nonempty(raw.get("sh"), raw.get("social_history")),
            "fh": _first_nonempty(raw.get("fh"), raw.get("family_history")),
            "vitals": raw.get("vitals"),
            "labs": raw.get("labs"),
            "imaging": raw.get("imaging"),
        }
        consult = {
            "chief_complaint": _first_nonempty(raw.get("chief_complaint"), raw.get("cc")),
            "history": _first_nonempty(raw.get("history"), raw.get("hpi")),
            "question": _first_nonempty(raw.get("question"), raw.get("consult_question"), raw.get("reason_for_consult")),
            "context": _first_nonempty(raw.get("context"), raw.get("background")),
            "referrer": raw.get("referrer"),
            "priority": raw.get("priority"),
        }

    # Final coercions → plain strings for stable prompting
    patient = {k: _as_str(v) for k, v in patient.items()}
    consult = {k: _as_str(v) for k, v in consult.items()}
    return {"patient": patient, "consult": consult}


# ------------------------------- summary ------------------------------------


def build_referral_summary(intake: Dict[str, Any], max_chars: int = 1600) -> str:
    """
    Build a compact, explicit referral summary from normalized intake.

    This is purposefully deterministic and only includes sections that exist to
    reduce noise. The caller (ai_core) may check len(summary) to decide whether
    to fall back to a minimal builder.
    """
    data = normalize_intake(intake)
    p = data["patient"]
    c = data["consult"]

    lines: List[str] = []

    # Patient header
    hdr_bits = []
    if p.get("age"):
        hdr_bits.append(str(p["age"]))
    if p.get("sex"):
        hdr_bits.append(str(p["sex"]))
    if hdr_bits:
        lines.append("Patient: " + ", ".join(hdr_bits))

    # Key question up front
    if c.get("question"):
        lines.append("Key question: " + c["question"])

    # Chief complaint / HPI
    if c.get("chief_complaint"):
        lines.append("Chief complaint: " + c["chief_complaint"])
    if c.get("history"):
        lines.append("Background: " + c["history"])

    # Past history / context
    if p.get("pmh"):
        lines.append("Past medical history: " + p["pmh"])
    if c.get("context"):
        lines.append("Context: " + c["context"])

    # Medications / allergies
    if p.get("medications"):
        lines.append("Medications: " + p["medications"])
    if p.get("allergies"):
        lines.append("Allergies: " + p["allergies"])

    # Objective data
    if p.get("vitals"):
        lines.append("Vitals: " + p["vitals"])
    if p.get("labs"):
        lines.append("Pertinent labs: " + p["labs"])
    if p.get("imaging"):
        lines.append("Imaging: " + p["imaging"])

    # Misc
    if p.get("fh"):
        lines.append("Family history: " + p["fh"])
    if p.get("sh"):
        lines.append("Social history: " + p["sh"])
    if c.get("referrer"):
        lines.append("Referrer: " + c["referrer"])
    if c.get("priority"):
        lines.append("Priority: " + c["priority"])

    summary = _strip_multiblank("\n".join([ln for ln in lines if ln]))
    if max_chars and len(summary) > max_chars:
        summary = summary[: max_chars - 3].rstrip() + "..."

    # Very light console log for validation
    print(f"[prompt_builder] Summary length={len(summary)} chars")

    return summary