Agents, Tasks, Crews: The Three Primitives

from crewai import Agent, LLM

agent = Agent(
    # Identity
    role="Senior Pharmacovigilance Analyst",          # What the agent IS
    goal="Identify and assess drug safety signals",   # What it's trying to ACHIEVE
    backstory=(                                        # Context shaping its behavior
        "You have 12 years of experience in pharmacovigilance at a major CRO. "
        "You are systematic, citation-focused, and flag any signal that exceeds "
        "a reporting threshold, even if causality is uncertain."
    ),

    # Model
    llm=LLM(model="gpt-4o"),                          # Which LLM to use (default: gpt-4)

    # Capabilities
    tools=[],                                          # List of tools the agent can call
    allow_delegation=False,                            # Can it delegate tasks to other agents?

    # Memory
    memory=True,                                       # Enable short + long-term memory

    # Execution
    max_iter=15,                                       # Max reasoning iterations before stopping
    max_rpm=None,                                      # Max requests per minute (rate limiting)
    max_execution_time=None,                           # Max seconds per task

    # Debugging
    verbose=True,                                      # Print reasoning steps
)

backstory="You are a helpful assistant."

backstory=(
    "You spent 8 years as a clinical research coordinator at an oncology center "
    "before moving into regulatory affairs. You have reviewed over 200 IND applications "
    "and are known for catching protocol deviations others miss. "
    "You write in precise regulatory language and always reference specific CFR sections."
)

from crewai import Task

task = Task(
    # What to do
    description=(
        "Analyze the adverse event reports for Drug X from Q1 2026. "
        "Focus on cardiovascular events. Identify any signal exceeding 2x "
        "the expected rate based on the drug label."
    ),

    # What done looks like
    expected_output=(
        "A structured report with three sections: "
        "1) Executive Summary (2-3 sentences), "
        "2) Signal Analysis (table with signal name, observed rate, expected rate, ROR), "
        "3) Recommendations (numbered list of next steps)."
    ),

    # Who does it
    agent=pharmacovigilance_analyst,

    # What previous task outputs to include as context
    context=[data_extraction_task],                    # Optional

    # Output options
    output_file="safety_report.md",                    # Optional: save to file
    output_pydantic=SafetyReport,                      # Optional: parse into Pydantic model
    output_json=True,                                  # Optional: force JSON output

    # Execution options
    async_execution=False,                             # Run concurrently with other tasks?

    # Callback (called when task completes)
    callback=on_task_complete,                         # Optional
)

expected_output="A good analysis."

expected_output=(
    "A numbered list of exactly 5 safety signals. "
    "Each signal must include: signal name, observed incidence rate (as a percentage), "
    "expected incidence rate from the drug label, the reporting odds ratio (ROR), "
    "and a one-sentence clinical interpretation. "
    "Format as a markdown table."
)

extraction_task = Task(
    description="Extract all adverse event records from the Q1 2026 safety database.",
    expected_output="A JSON array of adverse event records with fields: id, date, event_type, severity.",
    agent=data_engineer,
)

analysis_task = Task(
    description="Analyze the extracted adverse event records for safety signals.",
    expected_output="A safety signal report in markdown format.",
    agent=pharmacovigilance_analyst,
    context=[extraction_task],  # <-- output of extraction_task is injected here
)

from crewai import Crew, Process, LLM

crew = Crew(
    # Members and work
    agents=[data_engineer, analyst, writer, reviewer],
    tasks=[extraction_task, analysis_task, writing_task, review_task],

    # Execution strategy
    process=Process.sequential,                        # or Process.hierarchical

    # For hierarchical process only
    manager_llm=LLM(model="gpt-4o"),                  # LLM used by the manager agent
    manager_agent=None,                                # Or provide a custom manager Agent

    # Memory (shared across all agents in this crew)
    memory=False,                                      # Enable crew-level memory

    # Monitoring
    verbose=True,
    step_callback=on_step,                             # Called after each agent action
    task_callback=on_task_done,                        # Called after each task completes

    # Performance
    max_rpm=10,                                        # Rate limit across the entire crew
    share_crew=False,                                  # Share crew data with CrewAI platform
    output_log_file="crew_log.txt",                    # Log all output to file
)

# Synchronous
result = crew.kickoff()
print(result.raw)              # Final output as string
print(result.pydantic)         # Pydantic model if output_pydantic was set
print(result.json_dict)        # Dict if output_json was set
print(result.token_usage)      # Token consumption stats

# With input variables (interpolated into task descriptions)
result = crew.kickoff(inputs={"compound": "metformin", "quarter": "Q1 2026"})
# In task descriptions, use {compound} and {quarter} as placeholders

research_task = Task(
    description=(
        "Research the safety profile of {compound} using data from {quarter}. "
        "Focus on cardiovascular adverse events."
    ),
    expected_output="A structured safety summary.",
    agent=researcher,
)

crew = Crew(agents=[researcher], tasks=[research_task], process=Process.sequential)

# Inputs are interpolated at runtime
result = crew.kickoff(inputs={"compound": "semaglutide", "quarter": "Q1 2026"})

crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,
)

research_task → writing_task → review_task
     ↑                ↑              ↑
   agent=researcher  agent=writer  agent=reviewer

from crewai import LLM

crew = Crew(
    agents=[specialist_a, specialist_b, specialist_c],
    tasks=[task_1, task_2, task_3],
    process=Process.hierarchical,
    manager_llm=LLM(model="gpt-4o"),   # strong model recommended for manager
    verbose=True,
)

from crewai import Agent, Task, Crew, Process, LLM
from crewai_tools import SerperDevTool

# --- LLM ---
llm = LLM(model="gpt-4o")

# --- Tools ---
search = SerperDevTool()

# --- Agents ---
researcher = Agent(
    role="Clinical Research Analyst",
    goal="Find accurate, current clinical evidence for pharmaceutical questions",
    backstory=(
        "You are a clinical research analyst with expertise in systematic reviews. "
        "You evaluate evidence quality using GRADE criteria and flag low-quality evidence."
    ),
    tools=[search],
    llm=llm,
    verbose=True,
)

writer = Agent(
    role="Medical Communications Writer",
    goal="Translate clinical evidence into clear professional communications",
    backstory=(
        "You have 8 years of medical writing experience across pharma, biotech, and CROs. "
        "You write for HCP audiences and follow AMWA style guidelines."
    ),
    llm=llm,
    verbose=True,
)

reviewer = Agent(
    role="Regulatory Compliance Reviewer",
    goal="Ensure communications are accurate, balanced, and compliant",
    backstory=(
        "You are a regulatory affairs specialist with FDA review experience. "
        "You apply 21 CFR Part 202 standards and flag any promotional language "
        "that makes claims unsupported by the cited evidence."
    ),
    llm=llm,
    verbose=True,
)

# --- Tasks ---
research_task = Task(
    description=(
        "Research the clinical evidence for {drug} in treating {indication}. "
        "Include Phase 3 trial results, approved indications, and major safety findings."
    ),
    expected_output=(
        "A structured evidence summary with sections: "
        "Efficacy (key Phase 3 results with effect sizes), "
        "Safety (adverse events above 5% incidence), "
        "Regulatory Status (approved indications by region)."
    ),
    agent=researcher,
)

writing_task = Task(
    description=(
        "Write a 400-word HCP-facing medical communication on {drug} for {indication} "
        "based on the research summary."
    ),
    expected_output=(
        "A 400-word communication with: title, clinical context paragraph, "
        "efficacy section, safety section, and conclusion. "
        "No promotional language. All claims must reference the research summary."
    ),
    agent=writer,
    context=[research_task],
)

review_task = Task(
    description=(
        "Review the medical communication for accuracy, balance, and regulatory compliance. "
        "Apply 21 CFR Part 202 promotional material standards."
    ),
    expected_output=(
        "A review report with: "
        "Status (Approved / Approved with Changes / Rejected), "
        "Issues Found (numbered list, empty if none), "
        "Required Changes (numbered list, empty if approved)."
    ),
    agent=reviewer,
    context=[writing_task],
)

# --- Crew ---
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,
    verbose=True,
)

# --- Run ---
result = crew.kickoff(inputs={
    "drug": "tirzepatide",
    "indication": "type 2 diabetes",
})

print("=== FINAL REVIEW OUTPUT ===")
print(result.raw)