CrewAI Patterns#

Crew
├── Process: Sequential | Hierarchical
├── Agents
│   ├── Agent(role, goal, backstory, tools, llm)
│   └── Agent(...)
└── Tasks
    ├── Task(description, expected_output, agent)
    └── Task(description, expected_output, agent, context=[prior_task])

from crewai import Agent
import anthropic

# Verify current CrewAI integration patterns at crewai.com/docs
clinical_analyst = Agent(
    role="Clinical Criteria Analyst",
    goal=(
        "Evaluate whether a patient's clinical profile meets the criteria "
        "for a requested medical procedure based on current clinical guidelines."
    ),
    backstory=(
        "You are a clinical informatics specialist with deep expertise in "
        "evidence-based medicine and payer clinical criteria. You are precise, "
        "evidence-driven, and always cite the specific guideline section "
        "supporting your evaluation. You do not make diagnoses or treatment "
        "recommendations — you evaluate criteria only."
    ),
    tools=[guideline_search_tool, patient_data_tool],
    llm="claude-opus-4-8",  # Verify current model ID at docs.anthropic.com
    verbose=True,
    allow_delegation=False,  # This worker does not delegate further
)

from crewai import Task

clinical_eval_task = Task(
    description=(
        "Evaluate whether patient {patient_id} meets clinical criteria for "
        "procedure {procedure_code}. "
        "1. Review the patient summary provided in context. "
        "2. Search for the relevant clinical guidelines for this procedure. "
        "3. Evaluate each criterion. "
        "4. Produce a structured evaluation with: criteria_met (bool), "
        "   rationale (string), supporting_guidelines (list of citations)."
    ),
    expected_output=(
        "A JSON object with: criteria_met (boolean), rationale (string explaining "
        "the evaluation), supporting_guidelines (list of citation strings). "
        "Example: {\"criteria_met\": true, \"rationale\": \"...\", "
        "\"supporting_guidelines\": [\"ADA Standards 2025, Section 9\"]}"
    ),
    agent=clinical_analyst,
    context=[patient_retrieval_task],  # receives patient_retrieval_task output
)

from crewai import Crew, Process

prior_auth_crew = Crew(
    agents=[retrieval_agent, clinical_analyst, documentation_agent],
    tasks=[patient_retrieval_task, clinical_eval_task, documentation_task],
    process=Process.sequential,
    verbose=True,
)

result = prior_auth_crew.kickoff(inputs={
    "patient_id": "P-12345",
    "procedure_code": "95810",
})

"""
CrewAI sequential prior authorization workflow.
Educational Example — Illustrative multi-agent configuration.
Not intended for clinical decision making.

Prerequisites: pip install crewai anthropic
"""
from __future__ import annotations

try:
    from crewai import Agent, Task, Crew, Process
    from crewai.tools import BaseTool
    CREWAI_AVAILABLE = True
except ImportError:
    CREWAI_AVAILABLE = False
    print("CrewAI not installed. Run: pip install crewai")

from pydantic import BaseModel, Field
import json


# ── Tool definitions ─────────────────────────────────────────────────────────

class PatientDataInput(BaseModel):
    patient_id: str = Field(description="Patient identifier")

class GuidelineSearchInput(BaseModel):
    query: str = Field(description="Natural language search query for clinical guidelines")
    procedure_code: str = Field(default="", description="Optional CPT/HCPCS code to filter results")


if CREWAI_AVAILABLE:
    class GetPatientSummaryTool(BaseTool):
        name: str = "get_patient_summary"
        description: str = (
            "Retrieve a patient's clinical summary including active diagnoses, "
            "current medications, and relevant lab values. Use when you need "
            "patient-specific clinical context. Returns structured patient data."
        )
        args_schema: type[BaseModel] = PatientDataInput

        def _run(self, patient_id: str) -> str:
            # Stub: production calls FHIR R4 API
            data = {
                "patient_id": patient_id,
                "diagnoses": ["Obstructive Sleep Apnea (suspected)", "Obesity"],
                "bmi": 36.1,
                "epworth_score": 14,
                "ahi_screening": 22.3,
                "current_medications": ["Lisinopril 10mg", "Metformin 500mg"],
            }
            return json.dumps(data)

    class SearchGuidelinesTool(BaseTool):
        name: str = "search_clinical_guidelines"
        description: str = (
            "Search the institutional clinical guidelines library for evidence-based "
            "criteria and recommendations. Use when evaluating medical necessity or "
            "clinical appropriateness. Returns up to 3 relevant guideline passages."
        )
        args_schema: type[BaseModel] = GuidelineSearchInput

        def _run(self, query: str, procedure_code: str = "") -> str:
            # Stub: production queries vector store
            return json.dumps({
                "results": [
                    {
                        "title": "AASM Sleep Study Guidelines 2023",
                        "content": "Polysomnography (CPT 95810) is indicated for patients with "
                                  "suspected OSA with Epworth Sleepiness Scale ≥ 10 and AHI ≥ 15 "
                                  "on home sleep test, or BMI ≥ 35 with symptoms.",
                        "relevance": 0.91
                    }
                ]
            })

    # ── Agents ───────────────────────────────────────────────────────────────

    retrieval_agent = Agent(
        role="Clinical Data Retrieval Specialist",
        goal="Accurately retrieve complete patient clinical data required for prior authorization evaluation.",
        backstory=(
            "You are a clinical data specialist who retrieves structured patient information "
            "from the EHR system. You retrieve exactly the data requested, verify completeness, "
            "and present it in a structured format for downstream clinical evaluation."
        ),
        tools=[GetPatientSummaryTool()],
        verbose=True,
        allow_delegation=False,
    )

    clinical_analyst_agent = Agent(
        role="Clinical Criteria Analyst",
        goal="Evaluate clinical necessity using evidence-based guidelines. Produce clear, cited evaluations.",
        backstory=(
            "You are a clinical informatics expert specializing in medical necessity evaluation. "
            "You evaluate patient data against current evidence-based criteria, always citing "
            "specific guideline sections. You are objective and evidence-driven. "
            "You do not make diagnoses — you evaluate documented criteria only."
        ),
        tools=[SearchGuidelinesTool()],
        verbose=True,
        allow_delegation=False,
    )

    documentation_agent = Agent(
        role="Prior Authorization Documentation Specialist",
        goal="Draft clear, complete prior authorization letters based on clinical evaluation results.",
        backstory=(
            "You are a medical documentation specialist who translates clinical evaluations "
            "into well-structured prior authorization determination letters. "
            "Your letters are precise, professionally written, and clearly communicate "
            "the clinical rationale. All letters are clearly marked as drafts requiring physician review."
        ),
        tools=[],  # Documentation agent uses no tools — works from context only
        verbose=True,
        allow_delegation=False,
    )

    # ── Tasks ─────────────────────────────────────────────────────────────────

    retrieval_task = Task(
        description=(
            "Retrieve the clinical summary for patient {patient_id}. "
            "Use the get_patient_summary tool with the provided patient ID. "
            "Verify that the summary includes: diagnoses, BMI, relevant screening scores, "
            "and current medications. Present the complete summary."
        ),
        expected_output=(
            "A complete patient clinical summary including: active diagnoses, BMI, "
            "relevant screening test scores, and current medications. "
            "Format as a structured list."
        ),
        agent=retrieval_agent,
    )

    evaluation_task = Task(
        description=(
            "Evaluate whether the patient meets clinical criteria for procedure {procedure_code}. "
            "1. Review the patient summary from the previous task. "
            "2. Search for relevant clinical guidelines using the procedure code. "
            "3. Evaluate each criterion in the guidelines against the patient's data. "
            "4. Produce a structured determination: criteria_met (yes/no), specific criteria "
            "   met and not met, and the guideline citations supporting each point."
        ),
        expected_output=(
            "A clinical evaluation report containing: "
            "overall determination (MEETS CRITERIA / DOES NOT MEET CRITERIA), "
            "specific criterion-by-criterion analysis, "
            "and supporting guideline citations for each point."
        ),
        agent=clinical_analyst_agent,
        context=[retrieval_task],
    )

    documentation_task = Task(
        description=(
            "Draft a prior authorization determination letter based on the clinical evaluation. "
            "The letter must include: patient identifier, procedure requested, "
            "determination (approval/denial), clinical rationale with guideline citations, "
            "and a clear statement that this is a DRAFT requiring physician review. "
            "Label the letter: 'Educational Example — Not for clinical use.'"
        ),
        expected_output=(
            "A professionally formatted prior authorization determination letter draft. "
            "Must be clearly marked as DRAFT PENDING PHYSICIAN REVIEW. "
            "Must include clinical rationale with specific guideline citations."
        ),
        agent=documentation_agent,
        context=[retrieval_task, evaluation_task],
    )

    # ── Crew ─────────────────────────────────────────────────────────────────

    prior_auth_crew = Crew(
        agents=[retrieval_agent, clinical_analyst_agent, documentation_agent],
        tasks=[retrieval_task, evaluation_task, documentation_task],
        process=Process.sequential,
        verbose=True,
    )


def run_prior_auth_crew(patient_id: str, procedure_code: str) -> str:
    """Run the prior authorization crew for a given patient and procedure."""
    if not CREWAI_AVAILABLE:
        return "CrewAI not available. Install with: pip install crewai"

    result = prior_auth_crew.kickoff(inputs={
        "patient_id": patient_id,
        "procedure_code": procedure_code,
    })
    return str(result)


if __name__ == "__main__":
    print("=== CrewAI Prior Authorization Crew ===")
    print("Educational Example — Not for clinical use\n")

    output = run_prior_auth_crew("P-DEMO-001", "95810")
    print("\n=== Final Output ===")
    print(output)

# Hierarchical process — manager agent decomposes and delegates
hierarchical_crew = Crew(
    agents=[retrieval_agent, clinical_analyst_agent, documentation_agent],
    tasks=[retrieval_task, evaluation_task, documentation_task],
    process=Process.hierarchical,
    manager_llm="claude-opus-4-8",  # Manager uses frontier model; workers can use smaller models
    verbose=True,
)
# In hierarchical mode, the manager can re-delegate tasks and adapt the workflow
# based on interim results — suitable for complex, adaptive workflows.