CrewAI: a QA team in code

import { test, expect } from '@playwright/test';

const URL = 'https://app.thetestingacademy.com/ai/blueprint/learn/crewai.html';

test('the triage crew runs three tasks in order and passes context forward', async ({ page }) => {
  await page.goto(URL);
  await page.getByTestId('crew-kickoff').click();

  await expect(page.getByTestId('crew-status')).toContainText('3 of 3 tasks');
  await expect(page.getByTestId('crew-task-3')).toHaveAttribute('data-state', 'done');
  await expect(page.getByTestId('crew-task-3')).toContainText('context=[triage_task, root_cause_task]');
  await expect(page.getByTestId('crew-out-3')).toContainText('answered by groq');
});

test('agent 3 over the 8,000-token cap stops the crew when there is no fallback', async ({ page }) => {
  await page.goto(URL);
  await page.getByTestId('crew-max-3').fill('3300');
  await expect(page.getByTestId('crew-bar-3')).toContainText('8,100 of 8,000');

  await page.getByTestId('crew-kickoff').click();
  await expect(page.getByTestId('crew-task-2')).toHaveAttribute('data-state', 'done');
  await expect(page.getByTestId('crew-task-3')).toHaveAttribute('data-state', 'fail');
  await expect(page.getByTestId('crew-status')).toContainText('413');
});

test('with a DeepSeek key the FallbackLLM finishes the run', async ({ page }) => {
  await page.goto(URL);
  await page.getByTestId('crew-max-3').fill('3300');
  await page.getByTestId('crew-ds-key').check();
  await page.getByTestId('crew-kickoff').click();

  await expect(page.getByTestId('crew-log')).toContainText('falling back to secondary provider');
  await expect(page.getByTestId('crew-out-3')).toContainText('answered by deepseek');
  await expect(page.getByTestId('crew-status')).toContainText('3 of 3 tasks');
});

# ---------------------------------------------------------------------------
# Task 1: Bug Triage
# ---------------------------------------------------------------------------
triage_task = Task(
    description=f"""Triage this bug report from Jira:

{bug_report}

Work through your triage checklist and deliver:
1. SEVERITY: one of S0-S4, with the exact line of the report that justifies it.
2. PRIORITY: one of P0-P4, with the business reasoning (users affected, money at
   risk, workaround available or not). If severity and priority differ, explain why.
3. Do you AGREE with the reporter's severity and the Jira priority as filed? If not,
   state your override and the reason.
4. CATEGORY: exactly one from your taxonomy.
5. AFFECTED COMPONENT / MODULE and the most likely system layer.
6. BUSINESS IMPACT: what it costs if this ships as-is.
7. MISSING INFORMATION: what you would ask the reporter for, and the assumption you
   triaged under in the meantime.
8. CONFIDENCE: High / Medium / Low.""",

    expected_output="""A structured triage verdict in markdown with these headed
    sections: Severity, Priority, Reporter Disagreement, Category, Affected Component,
    Business Impact, Missing Information, Confidence. Each rating must quote the
    evidence line from the bug report that drove it.

    HARD LIMIT: 400 words. This verdict is handed to two more agents, so every
    word you spend costs them room. Use compact bullets, not wide tables.
    Finishing every section matters more than depth in any one of them.""",
    agent=bug_analyst,
)


# ---------------------------------------------------------------------------
# Task 2: Root Cause (uses triage output as context)
# ---------------------------------------------------------------------------
root_cause_task = Task(
    description=f"""Using the triage verdict from the previous task, investigate WHY
this bug happens:

{bug_report}

Deliver:
1. DIFFERENTIAL DIAGNOSIS: what is different between the working case and the
   failing case? Start here.
2. PRIMARY ROOT CAUSE HYPOTHESIS with a confidence percentage, the system layer it
   lives in, the specific function or config that is suspect, and the evidence line
   from the report supporting it.
3. TWO ALTERNATE HYPOTHESES, ranked, each with confidence and evidence.
4. KILL TEST for each of the three: the single fastest check that would disprove it.
5. LAYER VERDICT: frontend, backend, data, infra or third-party, and what the
   evidence lets you EXONERATE.
6. INVESTIGATION STEPS in order, naming the exact log file and grep string, the
   dashboard and metric, the SQL query, the DevTools network call, and the git
   command to diff the suspect release.
7. BLAST RADIUS: what else shares this code path and is probably broken too.
8. WHAT YOU COULD NOT DETERMINE and the one piece of data that would settle it.""",

    expected_output="""A root cause analysis in markdown with: Differential Diagnosis,
    Primary Hypothesis (with confidence % and kill test), Alternate Hypotheses 2 and 3,
    Layer Verdict including what is exonerated, ordered Investigation Steps with named
    artifacts, Blast Radius, and Open Questions.

    HARD LIMIT: 650 words. This analysis is handed to the test strategist, so
    every word you spend costs them room. Use compact bullets, not wide
    markdown tables. Finishing every section matters more than depth in any one
    of them: a complete short report beats a detailed one that gets cut off.""",
    agent=root_cause_agent,
    context=[triage_task],  # Receives output from triage
)


# ---------------------------------------------------------------------------
# Task 3: Test Recommendation (uses both previous outputs)
# ---------------------------------------------------------------------------
test_task = Task(
    description=f"""Using the triage verdict and the root cause analysis from the
previous tasks, design the test strategy for this bug:

{bug_report}

Deliver:
1. THE MISSING TEST: name the single test that should have existed and would have
   caught this before production, and explain why the current suite missed it.
2. VERIFICATION TEST: the one test that proves the fix works. State its level.
3. REGRESSION SET: 3-5 cases that stop this specific bug returning, each pinned to
   the cheapest level that can catch it (unit / API / E2E) with a one-line reason.
4. BOUNDARY AND EDGE CASES: apply boundary value analysis to the failing threshold
   in this report, plus the negative and state-transition cases.
5. PLAYWRIGHT TYPESCRIPT: runnable code for the E2E tests worth automating,
   following your standards (web-first assertions, no hard waits, role/testid
   locators, API-seeded state, route interception where it isolates the layer).
6. WHAT TO KEEP MANUAL and why.
7. EXIT CRITERIA: what must be green before this ticket is closed.""",

    expected_output="""A test strategy in markdown: the Missing Test, a table of tests
    (ID, Level, Type, What it proves, Priority), runnable Playwright TypeScript code
    blocks for the E2E cases, a Manual-only section with justification, and Exit
    Criteria.

    HARD LIMIT: 800 words including the code. You are the last agent and have
    the least room, so budget it: keep the tables narrow, write ONE complete
    Playwright spec file rather than several half-finished ones, and make sure
    you reach Exit Criteria. A complete short plan beats a truncated long one.""",
    agent=test_recommender,
    context=[triage_task, root_cause_task],  # Uses both outputs
)


# ---------------------------------------------------------------------------
# Step 3 + 4 - Assemble the Crew and kick it off
# ---------------------------------------------------------------------------
crew = Crew(
    agents=[bug_analyst, root_cause_agent, test_recommender],
    tasks=[triage_task, root_cause_task, test_task],
    process=Process.sequential,
    verbose=True,
)

class FallbackLLM(BaseLLM):
    """Try the primary LLM, fall back to the secondary when it fails.

    Why this exists: a free-tier key WILL let you down mid-demo. Rate limit
    (429), request too large (413), expired key (401), provider outage (5xx) -
    any of them kills the crew halfway through and you lose the whole run.
    This wrapper catches the failure on that one call and re-issues it against
    the backup provider, so the crew keeps going instead of crashing.

    CrewAI's `LLM` is a factory (it returns a provider-specific object), so you
    cannot subclass it. Wrap it instead and delegate.
    """

    primary: Any = None
    secondary: Any = None

    @staticmethod
    def _is_empty(response):
        """An empty string is a failure too, not a valid answer.

        DeepSeek occasionally returns empty content on long generations. That
        is NOT an exception, so a try/except alone never catches it: CrewAI
        just raises "Invalid response from LLM call - None or empty" a few
        steps later and the whole crew dies. Treat it as a failure here.
        """
        return response is None or (isinstance(response, str) and not response.strip())

    def call(self, *args, **kwargs):
        try:
            response = self.primary.call(*args, **kwargs)
            if not self._is_empty(response):
                return response
            reason = "returned an EMPTY response"
        except Exception as exc:
            reason = f"raised {type(exc).__name__}: {str(exc)[:150]}"
            if self.secondary is None:
                raise

        if self.secondary is None:
            raise RuntimeError(f"Primary LLM {reason} and no fallback is configured")

        print(f"\n[llm] PRIMARY {reason}\n[llm] falling back to secondary provider...\n")
        response = self.secondary.call(*args, **kwargs)
        if self._is_empty(response):
            raise RuntimeError("Both primary and fallback LLMs returned empty responses")
        return response

    # delegate capability checks to the primary
    def supports_function_calling(self):
        return self.primary.supports_function_calling()

    def supports_stop_words(self):
        return self.primary.supports_stop_words()

    def get_context_window_size(self):
        return self.primary.get_context_window_size()


def make_groq_llm(max_tokens):
    return LLM(
        model=f"openai/{os.getenv('GROQ_MODEL')}",
        api_key=os.getenv("GROQ_API_KEY"),
        base_url=os.getenv("GROQ_BASE_URL"),
        max_tokens=max_tokens,
        temperature=0.3,  # triage should be consistent, not creative
    )


def make_deepseek_llm(max_tokens=DEEPSEEK_MAX_TOKENS):
    return LLM(
        model=f"deepseek/{os.getenv('DEEPSEEK_MODEL', 'deepseek-v4-flash')}",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),
        max_tokens=max_tokens,
        temperature=0.3,
    )


# Which provider leads. Groq is fast, free and reliable here, so it leads. Its
# catch is the 8000-token request cap above, which is why every task also
# carries an explicit length limit: an agent that writes past its budget gets
# cut off mid-sentence. DeepSeek has no such cap but intermittently returns
# empty or short responses under long prompts, so it rides shotgun instead.
# Set LLM_PROVIDER=deepseek in .env to flip the order.
PRIMARY_PROVIDER = os.getenv("LLM_PROVIDER", "groq").lower()


def make_llm(groq_tokens, deepseek_tokens):
    """Build one LLM per agent, with the other provider wired in as a parachute.

    Each provider gets its own budget because their limits differ: Groq must
    stay under GROQ_TPM_LIMIT (prompt + output), DeepSeek can breathe.
    """
    has_groq = bool(os.getenv("GROQ_API_KEY"))
    has_deepseek = bool(os.getenv("DEEPSEEK_API_KEY"))

    groq = make_groq_llm(groq_tokens) if has_groq else None
    deepseek = make_deepseek_llm(deepseek_tokens) if has_deepseek else None

    if PRIMARY_PROVIDER == "groq":
        primary, secondary = groq, deepseek
    else:
        primary, secondary = deepseek, groq

    primary = primary or secondary
    if primary is None:
        raise RuntimeError("Set GROQ_API_KEY and/or DEEPSEEK_API_KEY in .env")
    if secondary is primary:
        secondary = None

    if secondary is None:
        return primary
    return FallbackLLM(
        model=f"{PRIMARY_PROVIDER}-with-fallback", primary=primary, secondary=secondary
    )


# Budgets: (groq, deepseek). The Groq numbers keep prompt + output under
# GROQ_TPM_LIMIT; the DeepSeek numbers are what these reports actually need.
triage_llm = make_llm(4000, 4000)  # smallest prompt, runs first
rca_llm = make_llm(4000, 5000)     # prompt also carries task 1's output
sdet_llm = make_llm(3200, 6000)    # prompt also carries task 1 + task 2 output

# Test Ananlyst Agent
# 
# a senior QA with 15 years (JIRA MD)
#  of experience. Based on the feature, 
# it will just analyze the requirement
# and suggest a 5-10 testcases(p0 testcases).

from crewai import Agent,Task, Crew
from crewai import LLM
from dotenv import load_dotenv
import os


# By Default crew AI actually the brain which
# OpenAI - GROQ API Key


# Step 0 - Set up the Brain
# Step 1. - Define the Agent (identity)
# Step 2. - Give the Task to the Agent
# Step 3. Add them to the Crew
# Step 4. Kick Off Agent.

# Prompt vs Skill vs AI Agent

# We need use the GROQ gpt-oss-120b model


# Step 0 - Set up the Brain (Groq LLM)
load_dotenv()  # reads the .env file in this folder

# Groq exposes an OpenAI-compatible API, so we use the "openai/" provider
# prefix with a custom base_url pointing at Groq.
# GROQ_MODEL in .env = openai/gpt-oss-120b (exact ID from groq.com console)
groq_llm = LLM(
    model=f"openai/{os.getenv('GROQ_MODEL')}",
    api_key=os.getenv("GROQ_API_KEY"),
    base_url=os.getenv("GROQ_BASE_URL"),
)

# Step 1. - Define the Agent (identity)
qa_agent = Agent(
    role="QA Enginner",
    goal="Analyse the feature or the requirements, and create 5-10 test cases out of it.",
    backstory="You are a senior QA engineer with 15 years of experience in test planning and testcases creation",
    llm = groq_llm,
    verbose=True
)

# Step 2 - Give the Task to the Agent
test_case_task = Task(
    description="Create 5-10 test cases",
    expected_output="A numbered list of 5-10 test cases with brief descriptions for a app.vwo.com Login page with the username, password and submit button with remember me functionality",
    agent=qa_agent
)

# Step 3. Add them to the Crew
crew = Crew(
    agents=[qa_agent],
    tasks=[test_case_task],
    verbose=True
)

# Step 4. kickOff
if __name__ == "__main__":
    result = crew.kickoff()
    print(result)

bug_analyst = Agent(
    role="Senior Bug Triage Analyst and Defect Classification Specialist",

    goal=(
        "Read the raw bug report and return one defensible triage verdict for it: "
        "exactly one SEVERITY, exactly one PRIORITY, and exactly one CATEGORY. "
        "Every one of the three must be justified by quoting the specific line of "
        "the bug report that drove the decision. Never guess numbers that are not "
        "in the report, and always call out what information is missing that would "
        "change your verdict."
    ),

    backstory="""You are a veteran QA engineer with 15+ years of experience. You have
    personally triaged well over 20,000 defects across e-commerce checkouts, payment
    gateways, and B2B SaaS dashboards, and you have run the daily bug triage meeting
    for teams of 30+ engineers. Triage is not paperwork to you. A wrong severity either
    wakes an on-call engineer at 2 AM for a typo, or lets a money-losing bug sit in the
    backlog for three sprints. You take it seriously.

    ## THE ONE RULE PEOPLE ALWAYS GET WRONG
    Severity and Priority are NOT the same thing, and you never collapse them:
    - SEVERITY = technical impact on the system. How badly is it broken? Set by QA.
      This is objective and does not care about the release calendar.
    - PRIORITY = business urgency. How soon must it be fixed relative to everything
      else? Driven by users affected, money at risk, and whether a workaround exists.
    A typo in the company name on the homepage is LOW severity but HIGH priority.
    A crash in an admin tool used by two internal people is HIGH severity but LOW
    priority. You explain this distinction whenever the two ratings differ.

    ## SEVERITY SCALE (technical impact)
    - S0 Blocker  : System down, data loss or data corruption, security breach,
                    money computed wrong, complete checkout/login failure. Nothing
                    can proceed.
    - S1 Critical : Major feature completely broken with NO workaround. Core user
                    journey blocked for a large segment.
    - S2 Major    : Feature impaired or partially wrong, but a reasonable workaround
                    exists. Journey is painful, not blocked.
    - S3 Minor    : Cosmetic, layout, wording, or minor inconvenience. Function is
                    intact.
    - S4 Trivial  : Typo, alignment nitpick, enhancement request, "would be nice".

    ## PRIORITY SCALE (business urgency and fix order)
    - P0 : Fix now, hotfix today, stop other work. Production + revenue or security.
    - P1 : Fix in the current sprint, before the next release ships.
    - P2 : Schedule into the next sprint. Normal backlog flow.
    - P3 : Fix when the area is touched next. Opportunistic.
    - P4 : Backlog / icebox. May legitimately never be fixed.

    ## CATEGORY TAXONOMY (pick exactly one, the closest root fit)
    Functional Logic, Data and Calculation, UI/UX and Layout, Performance,
    Security, API and Integration, Compatibility (browser/OS/device),
    Configuration and Deployment, Regression, Usability and Content.

    ## HOW YOU ACTUALLY DECIDE (your triage checklist)
    1. Environment first. Production outweighs staging outweighs local dev.
    2. Blast radius. All users, one segment, or one account? If the report does not
       say, you say so instead of inventing a percentage.
    3. Money and data path. Anything touching price, total, discount, tax, payment,
       or stored customer data starts at S0/S1 by default, even when the symptom
       looks cosmetic. A wrong total is never "just a display issue".
    4. Workaround. If a real workaround exists, severity drops one level. If it does
       not, priority rises one level.
    5. Reproducibility. Consistent and 100% reproducible beats intermittent. An
       intermittent bug is NOT automatically lower severity, it is often harder and
       you say so.
    6. Regression signal. "It worked before the last deploy" is a strong escalator.
       A freshly shipped regression is more urgent than an old known defect.
    7. Layer isolation. If the API response is correct but the UI shows something
       else, the defect is frontend/presentation, not backend. If the UI input is
       correct but the stored value is wrong, it is backend/data. Say which layer.
    8. Security and compliance. Any auth bypass, data exposure, PII leak, or
       injection vector is S0/P0 regardless of how few users hit it.

    ## RULES OF YOUR DESK
    - You NEVER inflate severity to get attention, and you never deflate it to
      protect a release date.
    - The reporter's severity is an input, not an instruction. If the evidence does
      not support it, you override it and state plainly why you disagreed.
    - You never invent facts. If user impact, error logs, or affected version are
      missing, you list them under "Missing Information" and state the assumption
      you triaged under.
    - You state a confidence level (High / Medium / Low) on your verdict. Low
      confidence with a clear list of what you need beats a confident guess.
    - You write for two audiences at once: an engineer who needs the technical
      signal, and a product manager who needs to know if it can wait.""",

    llm=triage_llm,
    verbose=True,
    allow_delegation=False,  # This agent handles its own work
)