Build an AI Business Intelligence Crew

Build a 4-agent AI crew that delivers ML-backed business proposals.

Introduction

30 Second Summary

When consulting firms analyze a company's data, they don't send one generalist. They deploy a team of specialists. A data engineer cleans the numbers, an analyst finds patterns, a strategist maps findings to solutions, and an advisor packages it all for the C-suite. Each person brings a unique lens that makes the final recommendation more trustworthy.

In this project, you will build that same team structure using AI agents. Four specialized agents collaborate to analyze business metrics, run machine learning tools, and produce a structured executive proposal with human approval before delivery.

What You'll Build

A production-grade AI Business Intelligence system where a crew of four CrewAI agents collaborates to produce an AI Transformation Proposal backed by scikit-learn ML evidence, real-time cost tracking, and MLflow tracing.

By the end of this project, you'll have:

  • A multi-agent crew with four data science roles (Data Engineer, ML Analyst, AI Strategist, Executive Advisor) that collaborate sequentially to produce business recommendations.
  • Custom ML tools for anomaly detection, customer segmentation, and churn prediction exposed to agents through CrewAI's @tool decorator.
  • Production guardrails, LLM hooks for real-time cost tracking, MLflow tracing for observability, and a human-in-the-loop approval gate on the final executive proposal.
  • Secret Mission: Add an AI Ethics Auditor agent that scores every recommendation for fairness, transparency, and accountability before the final proposal is delivered.

Are there any prerequisites?

This is an advanced project. You should be comfortable writing Python (including Pydantic models and basic async patterns), have an OpenAI API key ready, and have VS Code installed on your machine.

Set Up Your Environment

Before you can build your multi-agent AI system, you need a clean development environment with the right tools installed.

In this step, you will install uv (a fast Python package manager), create a virtual environment, install all project dependencies, and confirm everything is connected.

In this step, get ready to:
  • Install uv and create a Python 3.11 virtual environment.
  • Install CrewAI, scikit-learn, MLflow, and supporting packages.
  • Configure environment variables and start the MLflow tracking server.
Install uv and create a virtual environment

Your project needs isolated package management so dependencies do not conflict with your system Python. uv is an ultrafast Python package manager that replaces pip and venv in one tool.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Install uv by running this command:

macOS

curl -LsSf https://astral.sh/uv/install.sh | sh

Windows

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

What is uv?

uv is a Rust-based Python package manager from Astral. It handles virtual environments, dependency resolution, and package installation significantly faster than pip.

You will use it throughout this project to manage your packages.

  • Close and reopen your terminal so the uv command is available on your PATH.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir ai-bi-crew
cd ai-bi-crew
  • Create a virtual environment pinned to Python 3.11 by running:
uv venv --python 3.11

What does this command do?

uv venv --python 3.11 creates a .venv folder inside your project directory containing an isolated Python 3.11 interpreter.

Packages you install here stay separate from your system Python. If Python 3.11 is not already on your machine, uv downloads it automatically.

  • Activate the virtual environment by running:

macOS

source .venv/bin/activate

Windows

.venv\Scripts\activate

You should see (.venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.

  • Install all project dependencies by running:
uv pip install crewai scikit-learn mlflow tiktoken python-dotenv

What are these packages?

  • crewai is the multi-agent orchestration framework you will use to define agents and tasks.
  • scikit-learn provides the ML algorithms (Isolation Forest, K-Means, Logistic Regression) your agents will use as tools.
  • mlflow gives you a local dashboard for tracing agent decisions and tracking costs.
  • tiktoken counts tokens so you can estimate OpenAI API costs per call.
  • python-dotenv loads environment variables from a .env file so your API key stays out of your code.

The install takes a minute or two. You should see a list of successfully installed packages when it finishes.

Seeing a resolution error?

  • Make sure (.venv) appears in your terminal prompt. If not, re-run the activate command above.
  • If Python 3.11 was not found during venv creation, run uv venv --python 3.11 again. uv downloads Python 3.11 automatically if it is missing.

Still stuck? Help me troubleshoot my uv installation or virtual environment setup.

Configure environment variables and verify connectivity

Your agents need an OpenAI API key to call gpt-4o-mini. Storing it in a .env file keeps it out of your source code.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Visual Studio Code and press Enter to open it.
  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Select the ai-bi-crew folder on your Desktop, then click Open.
  • Create a new file called .env in the root of ai-bi-crew by clicking the new file icon in the Explorer sidebar.
  • Add the following content to .env:
OPENAI_API_KEY=your-api-key-here
OPENAI_MODEL_NAME=gpt-4o-mini
  • Replace your-api-key-here with your actual OpenAI API key (starts with sk-).
  • Save .env by pressing Cmd+S (macOS) or Ctrl+S (Windows).

Why two environment variables?

OPENAI_API_KEY authenticates your requests to OpenAI. OPENAI_MODEL_NAME tells CrewAI which model to use by default.

Setting gpt-4o-mini keeps costs low at $0.15 per million input tokens and $0.60 per million output tokens.

  • Verify CrewAI is installed correctly by running this in your terminal:
python -c "from crewai import Agent; print('CrewAI ready')"

You should see CrewAI ready printed. This confirms the package imported successfully.

  • Verify OpenAI connectivity by running:
python -c "from openai import OpenAI; from dotenv import load_dotenv; load_dotenv(); client = OpenAI(); r = client.chat.completions.create(model='gpt-4o-mini', messages=[{'role':'user','content':'Say hello'}]); print(r.choices[0].message.content)"

You should see a short greeting from the model. This confirms your API key is valid and the connection works.

Seeing an authentication error?

  • Open .env and check that your key starts with sk- and has no extra spaces or quotes around it.
  • Confirm you saved the file after pasting your key.
  • Make sure (.venv) is still active in your terminal. If you opened a new terminal window, reactivate the environment.

Still stuck? Help me debug my OpenAI API key connection in my CrewAI project.

Start the MLflow tracking server

MLflow provides a local dashboard where you can trace every decision your agents make, every tool call, and every token spent.

You will start the server now and leave it running in the background for the rest of the project.

  • Open a second terminal tab in VS Code by clicking Terminal in the menu bar, then clicking New Terminal.
  • Activate the virtual environment in this new terminal:

macOS

cd ~/Desktop/ai-bi-crew
source .venv/bin/activate

Windows

cd ~\Desktop\ai-bi-crew
.venv\Scripts\activate
  • Start the MLflow tracking server by running:
mlflow server

What does this command do?

mlflow server launches a local web server that records experiment data (agent traces, token counts, latency). It runs on http://127.0.0.1:5000 by default.

Leave this terminal running for the rest of the project. You will use your first terminal for all other commands.

  • Open your browser and go to http://127.0.0.1:5000.

You should see the MLflow dashboard with an empty experiments list. This confirms the server is running and ready to receive traces from your agents.

Dashboard not loading?

  • Check that (.venv) is active in the terminal where you ran mlflow server. If not, activate the environment and retry.
  • If port 5000 is already in use, stop the other process occupying it before running mlflow server again.
  • Verify mlflow installed by running python -c "import mlflow; print(mlflow.__version__)" in your first terminal.

Still stuck? Help me get my MLflow tracking server running.

✔️ Awesome, I've got everything!

Great. Double check you have saved your .env file and that both terminals are active (one for commands, one running mlflow server).

ⓧ I'd like to double check the full code

Here is the complete .env file you should have at this point:

OPENAI_API_KEY=your-api-key-here
OPENAI_MODEL_NAME=gpt-4o-mini

Your environment is ready. Next up, you will build the custom ML tools that give your agents statistical superpowers.

Build the ML Analytics Toolkit

Your development environment is ready. Now it's time to build the analytical engine that powers your AI crew.

Real data science teams don't just ask LLMs to "analyze things." They back every claim with statistical evidence. In this step, you'll create ML tools that give your agents the ability to detect anomalies, segment customers, and predict churn using established algorithms from scikit-learn.

In this step, get ready to:
  • Build a simulated data generator that produces 12 months of business metrics with injected anomalies.
  • Create three ML analysis tools (Isolation Forest, K-Means, Logistic Regression) decorated for CrewAI agent use.
  • Add an ethical bias checker and define Pydantic output models for structured results.
Build the simulated data generator

Your ML tools need data to analyze. Instead of requiring external datasets, you'll create a data generator that produces realistic company metrics for any industry.

  • In VS Code, create a new file called data_generator.py in your project folder.
  • Add the following code to build the business data generator:
import numpy as np
from typing import Dict, Any


def generate_business_data(industry: str, company_name: str) -> Dict[str, Any]:
    """Generate simulated business metrics for analysis."""
    np.random.seed(42)

    base_revenue = np.random.uniform(5_000_000, 50_000_000)
    months = 12

    revenue_trend = base_revenue + np.cumsum(
        np.random.normal(50_000, 200_000, months)
    )
    customer_counts = np.random.randint(1000, 50000, months)
    satisfaction_scores = np.random.uniform(3.2, 4.8, months)
    operational_costs = revenue_trend * np.random.uniform(0.6, 0.85, months)
    churn_rates = np.random.uniform(0.02, 0.15, months)

    # Inject anomalies for detection
    revenue_trend[3] = revenue_trend[3] * 0.4  # Revenue drop
    revenue_trend[8] = revenue_trend[8] * 1.8  # Revenue spike
    churn_rates[5] = 0.35  # Churn spike

    return {
        "company_name": company_name,
        "industry": industry,
        "monthly_revenue": revenue_trend.tolist(),
        "customer_counts": customer_counts.tolist(),
        "satisfaction_scores": satisfaction_scores.tolist(),
        "operational_costs": operational_costs.tolist(),
        "churn_rates": churn_rates.tolist(),
        "total_customers": int(customer_counts.mean()),
        "avg_satisfaction": float(satisfaction_scores.mean()),
        "avg_churn_rate": float(churn_rates.mean()),
    }

What does this code do?

This function generates 12 months of simulated business data using NumPy with a fixed random seed (42) for reproducibility. Every time your agents call this function, they get identical data to analyze.

Three anomalies are deliberately injected: a 60% revenue drop in month 4, a revenue spike in month 9, and a churn spike in month 6. These give the Isolation Forest tool something meaningful to detect.

  • Save data_generator.py.
  • Verify the function works by running this in your terminal:
python -c "from data_generator import generate_business_data; d = generate_business_data('Healthcare', 'Test'); print(f'Avg churn: {d[\"avg_churn_rate\"]:.2%}')"

You should see Avg churn: followed by a percentage value printed in your terminal.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is still activated (you should see the env name in your terminal prompt). Also confirm you saved data_generator.py in the same folder where your terminal is running.

help me fix a ModuleNotFoundError when importing data_generator

Create the custom ML tools

CrewAI agents call tools during their reasoning process. Each tool you build here wraps a scikit-learn algorithm and returns a human-readable string of findings that the agent can use in its response.

  • Create a new file called ml_tools.py in your project folder.
  • Add the imports and the first tool, detect_anomalies, which uses Isolation Forest to find unusual months in business data:
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.cluster import KMeans
from sklearn.linear_model import LogisticRegression
from crewai.tools import tool

from data_generator import generate_business_data


@tool("detect_anomalies")
def detect_anomalies(industry: str, company_name: str) -> str:
    """Detect anomalies in business revenue and operations data using Isolation Forest.
    Returns identified anomalous months with severity scores."""
    data = generate_business_data(industry, company_name)
    features = np.array(
        [data["monthly_revenue"], data["operational_costs"], data["churn_rates"]]
    ).T

    model = IsolationForest(contamination=0.2, random_state=42)
    predictions = model.fit_predict(features)
    scores = model.decision_function(features)

    anomalies = []
    for i, (pred, score) in enumerate(zip(predictions, scores)):
        if pred == -1:
            anomalies.append(
                f"Month {i+1}: anomaly score={score:.3f}, "
                f"revenue=${data['monthly_revenue'][i]:,.0f}, "
                f"churn={data['churn_rates'][i]:.1%}"
            )

    return (
        f"Isolation Forest Analysis for {company_name} ({industry}):\n"
        f"Analyzed 12 months of data across 3 features.\n"
        f"Found {len(anomalies)} anomalous periods:\n"
        + "\n".join(anomalies)
        + f"\nModel contamination parameter: 0.2"
    )

What does this code do?

The @tool decorator from CrewAI registers this function so agents can call it during execution. The function's docstring becomes the tool description that agents read to decide when to use it.

Isolation Forest is an unsupervised anomaly detection algorithm. It isolates outliers by randomly partitioning data. Points that require fewer partitions to isolate are more anomalous. The contamination=0.2 parameter tells the model to expect roughly 20% of data points to be anomalous.

  • Below the detect_anomalies function, add the customer segmentation tool using K-Means clustering:
@tool("segment_customers")
def segment_customers(industry: str, company_name: str) -> str:
    """Segment customers using K-Means clustering on behavioral metrics.
    Returns cluster profiles with business characteristics."""
    data = generate_business_data(industry, company_name)
    np.random.seed(42)

    n_customers = 200
    features = np.column_stack([
        np.random.exponential(500, n_customers),  # spend
        np.random.poisson(12, n_customers),  # frequency
        np.random.uniform(0, 1, n_customers),  # recency_score
    ])

    model = KMeans(n_clusters=4, random_state=42, n_init=10)
    labels = model.fit_predict(features)

    segments = []
    segment_names = ["High-Value Loyalists", "At-Risk Big Spenders",
                     "Occasional Buyers", "New/Dormant"]
    for i in range(4):
        mask = labels == i
        count = mask.sum()
        avg_spend = features[mask, 0].mean()
        avg_freq = features[mask, 1].mean()
        segments.append(
            f"  Segment '{segment_names[i]}': {count} customers, "
            f"avg spend=${avg_spend:.0f}, avg frequency={avg_freq:.1f}/year"
        )

    return (
        f"K-Means Customer Segmentation for {company_name} ({industry}):\n"
        f"Clustered {n_customers} customers into 4 segments:\n"
        + "\n".join(segments)
        + f"\nInertia: {model.inertia_:.0f}"
    )

What does this code do?

K-Means groups 200 simulated customers into 4 clusters based on spending, purchase frequency, and recency. The algorithm minimizes within-cluster variance, producing distinct customer personas the AI Strategist agent will later use to target recommendations.

The inertia_ metric measures how tightly packed each cluster is. Lower inertia means more distinct segments.

  • Below segment_customers, add the churn prediction tool using Logistic Regression:
@tool("predict_churn_risk")
def predict_churn_risk(industry: str, company_name: str) -> str:
    """Predict customer churn risk using Logistic Regression.
    Returns risk distribution and key predictive factors."""
    np.random.seed(42)
    n_samples = 300

    tenure = np.random.exponential(24, n_samples)
    monthly_spend = np.random.exponential(200, n_samples)
    support_tickets = np.random.poisson(2, n_samples)
    satisfaction = np.random.uniform(1, 5, n_samples)

    churn_probability = 1 / (1 + np.exp(
        -(0.5 - 0.02 * tenure - 0.001 * monthly_spend
          + 0.3 * support_tickets - 0.4 * satisfaction)
    ))
    churned = (churn_probability > 0.5).astype(int)

    X = np.column_stack([tenure, monthly_spend, support_tickets, satisfaction])
    model = LogisticRegression(random_state=42, max_iter=1000)
    model.fit(X, churned)

    feature_names = ["tenure", "monthly_spend", "support_tickets", "satisfaction"]
    coefficients = dict(zip(feature_names, model.coef_[0]))

    high_risk = (model.predict_proba(X)[:, 1] > 0.7).sum()
    medium_risk = ((model.predict_proba(X)[:, 1] > 0.4) &
                   (model.predict_proba(X)[:, 1] <= 0.7)).sum()
    low_risk = (model.predict_proba(X)[:, 1] <= 0.4).sum()

    return (
        f"Churn Prediction for {company_name} ({industry}):\n"
        f"Logistic Regression trained on {n_samples} customer records.\n"
        f"Model accuracy: {model.score(X, churned):.1%}\n"
        f"Risk distribution: High={high_risk}, Medium={medium_risk}, Low={low_risk}\n"
        f"Key predictors (coefficients):\n"
        + "\n".join(f"  {k}: {v:.4f}" for k, v in sorted(
            coefficients.items(), key=lambda x: abs(x[1]), reverse=True
        ))
        + f"\nChurn rate in dataset: {churned.mean():.1%}"
    )

What does this code do?

Logistic Regression is a supervised classifier that estimates the probability of customer churn based on four features: tenure, monthly spend, support tickets, and satisfaction. The model trains on 300 synthetic records with a known churn formula baked in.

The output breaks customers into High, Medium, and Low risk buckets and reports which features most strongly predict churn (via coefficients). This gives the Executive Advisor agent concrete numbers to reference in the final proposal.

  • Save ml_tools.py. You won't run it yet. The next substep adds the final tool to this file.
Add the ethical bias checker and define output models

Production AI systems need ethical guardrails. You'll add a tool that screens recommendations for language that could signal harmful bias, and then define the structured output schema for the crew's final deliverable.

  • In ml_tools.py, add the check_ethical_bias tool below the predict_churn_risk function:
@tool("check_ethical_bias")
def check_ethical_bias(recommendation: str) -> str:
    """Screen a recommendation for potential ethical concerns including
    demographic fairness, transparency, and accountability gaps."""
    concerns = []
    recommendation_lower = recommendation.lower()

    bias_keywords = ["automate layoffs", "replace workers", "eliminate jobs",
                     "surveillance", "monitor employees", "track behavior"]
    for keyword in bias_keywords:
        if keyword in recommendation_lower:
            concerns.append(
                f"FLAGGED: '{keyword}' detected. Ensure human oversight "
                f"and impact assessment are included."
            )

    if "explain" not in recommendation_lower and "transparent" not in recommendation_lower:
        concerns.append(
            "ADVISORY: Recommendation lacks explicit transparency mechanism. "
            "Consider adding explainability requirements."
        )

    if not concerns:
        return "PASSED: No ethical concerns detected. Recommendation appears balanced."

    return (
        f"ETHICAL REVIEW ({len(concerns)} items):\n"
        + "\n".join(f"  - {c}" for c in concerns)
        + "\nACTION: Address flagged items before finalizing proposal."
    )

What does this code do?

This tool performs keyword-based ethical screening. It checks recommendations against a list of concerning phrases (like "automate layoffs" or "surveillance") and flags them for human oversight.

It also checks whether the recommendation mentions transparency or explainability. If neither appears, it issues an advisory. This layered approach mirrors how production responsible AI systems operate.

  • Save ml_tools.py.
  • Verify the tools load correctly by running:
python -c "from ml_tools import detect_anomalies, segment_customers, predict_churn_risk, check_ethical_bias; print('All 4 tools loaded successfully')"

You should see All 4 tools loaded successfully printed in your terminal.

Seeing an import error?

If you see ModuleNotFoundError: No module named 'data_generator', confirm that data_generator.py and ml_tools.py are in the same folder. If you see an error about sklearn, make sure your virtual environment is activated.

help me debug import errors in my CrewAI ML tools file

✔️ Awesome, I've got everything!

Great. Double check you've saved ml_tools.py with all four tool functions.

ⓧ I'd like to double check the full code

import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.cluster import KMeans
from sklearn.linear_model import LogisticRegression
from crewai.tools import tool

from data_generator import generate_business_data


@tool("detect_anomalies")
def detect_anomalies(industry: str, company_name: str) -> str:
    """Detect anomalies in business revenue and operations data using Isolation Forest.
    Returns identified anomalous months with severity scores."""
    data = generate_business_data(industry, company_name)
    features = np.array(
        [data["monthly_revenue"], data["operational_costs"], data["churn_rates"]]
    ).T

    model = IsolationForest(contamination=0.2, random_state=42)
    predictions = model.fit_predict(features)
    scores = model.decision_function(features)

    anomalies = []
    for i, (pred, score) in enumerate(zip(predictions, scores)):
        if pred == -1:
            anomalies.append(
                f"Month {i+1}: anomaly score={score:.3f}, "
                f"revenue=${data['monthly_revenue'][i]:,.0f}, "
                f"churn={data['churn_rates'][i]:.1%}"
            )

    return (
        f"Isolation Forest Analysis for {company_name} ({industry}):\n"
        f"Analyzed 12 months of data across 3 features.\n"
        f"Found {len(anomalies)} anomalous periods:\n"
        + "\n".join(anomalies)
        + f"\nModel contamination parameter: 0.2"
    )


@tool("segment_customers")
def segment_customers(industry: str, company_name: str) -> str:
    """Segment customers using K-Means clustering on behavioral metrics.
    Returns cluster profiles with business characteristics."""
    data = generate_business_data(industry, company_name)
    np.random.seed(42)

    n_customers = 200
    features = np.column_stack([
        np.random.exponential(500, n_customers),  # spend
        np.random.poisson(12, n_customers),  # frequency
        np.random.uniform(0, 1, n_customers),  # recency_score
    ])

    model = KMeans(n_clusters=4, random_state=42, n_init=10)
    labels = model.fit_predict(features)

    segments = []
    segment_names = ["High-Value Loyalists", "At-Risk Big Spenders",
                     "Occasional Buyers", "New/Dormant"]
    for i in range(4):
        mask = labels == i
        count = mask.sum()
        avg_spend = features[mask, 0].mean()
        avg_freq = features[mask, 1].mean()
        segments.append(
            f"  Segment '{segment_names[i]}': {count} customers, "
            f"avg spend=${avg_spend:.0f}, avg frequency={avg_freq:.1f}/year"
        )

    return (
        f"K-Means Customer Segmentation for {company_name} ({industry}):\n"
        f"Clustered {n_customers} customers into 4 segments:\n"
        + "\n".join(segments)
        + f"\nInertia: {model.inertia_:.0f}"
    )


@tool("predict_churn_risk")
def predict_churn_risk(industry: str, company_name: str) -> str:
    """Predict customer churn risk using Logistic Regression.
    Returns risk distribution and key predictive factors."""
    np.random.seed(42)
    n_samples = 300

    tenure = np.random.exponential(24, n_samples)
    monthly_spend = np.random.exponential(200, n_samples)
    support_tickets = np.random.poisson(2, n_samples)
    satisfaction = np.random.uniform(1, 5, n_samples)

    churn_probability = 1 / (1 + np.exp(
        -(0.5 - 0.02 * tenure - 0.001 * monthly_spend
          + 0.3 * support_tickets - 0.4 * satisfaction)
    ))
    churned = (churn_probability > 0.5).astype(int)

    X = np.column_stack([tenure, monthly_spend, support_tickets, satisfaction])
    model = LogisticRegression(random_state=42, max_iter=1000)
    model.fit(X, churned)

    feature_names = ["tenure", "monthly_spend", "support_tickets", "satisfaction"]
    coefficients = dict(zip(feature_names, model.coef_[0]))

    high_risk = (model.predict_proba(X)[:, 1] > 0.7).sum()
    medium_risk = ((model.predict_proba(X)[:, 1] > 0.4) &
                   (model.predict_proba(X)[:, 1] <= 0.7)).sum()
    low_risk = (model.predict_proba(X)[:, 1] <= 0.4).sum()

    return (
        f"Churn Prediction for {company_name} ({industry}):\n"
        f"Logistic Regression trained on {n_samples} customer records.\n"
        f"Model accuracy: {model.score(X, churned):.1%}\n"
        f"Risk distribution: High={high_risk}, Medium={medium_risk}, Low={low_risk}\n"
        f"Key predictors (coefficients):\n"
        + "\n".join(f"  {k}: {v:.4f}" for k, v in sorted(
            coefficients.items(), key=lambda x: abs(x[1]), reverse=True
        ))
        + f"\nChurn rate in dataset: {churned.mean():.1%}"
    )


@tool("check_ethical_bias")
def check_ethical_bias(recommendation: str) -> str:
    """Screen a recommendation for potential ethical concerns including
    demographic fairness, transparency, and accountability gaps."""
    concerns = []
    recommendation_lower = recommendation.lower()

    bias_keywords = ["automate layoffs", "replace workers", "eliminate jobs",
                     "surveillance", "monitor employees", "track behavior"]
    for keyword in bias_keywords:
        if keyword in recommendation_lower:
            concerns.append(
                f"FLAGGED: '{keyword}' detected. Ensure human oversight "
                f"and impact assessment are included."
            )

    if "explain" not in recommendation_lower and "transparent" not in recommendation_lower:
        concerns.append(
            "ADVISORY: Recommendation lacks explicit transparency mechanism. "
            "Consider adding explainability requirements."
        )

    if not concerns:
        return "PASSED: No ethical concerns detected. Recommendation appears balanced."

    return (
        f"ETHICAL REVIEW ({len(concerns)} items):\n"
        + "\n".join(f"  - {c}" for c in concerns)
        + "\nACTION: Address flagged items before finalizing proposal."
    )

Now you'll define the structured output models. These Pydantic classes define the exact shape of the crew's final deliverable, so the Executive Advisor agent produces machine-parseable output instead of freeform text.

  • Create a new file called models.py in your project folder.
  • Add the following Pydantic models that define the proposal structure:
from pydantic import BaseModel, Field
from typing import List


class MLFinding(BaseModel):
    technique: str = Field(description="ML technique used")
    insight: str = Field(description="Key finding from the analysis")
    confidence: float = Field(description="Confidence score 0-1")
    evidence: str = Field(description="Numeric evidence supporting the finding")


class AIOpportunity(BaseModel):
    area: str = Field(description="Business area for AI intervention")
    solution: str = Field(description="Proposed GenAI/agentic solution")
    expected_impact: str = Field(description="Quantified expected impact")
    risk_benefit: str = Field(description="Ethical risk/benefit statement")


class ROIProjection(BaseModel):
    cost_savings_annual: float = Field(description="Projected annual cost savings in USD")
    revenue_uplift_percent: float = Field(description="Projected revenue increase percentage")
    customer_satisfaction_delta: float = Field(description="Expected CSAT improvement points")
    payback_period_months: int = Field(description="Months to positive ROI")


class BusinessProposal(BaseModel):
    company_name: str = Field(description="Target company name")
    industry: str = Field(description="Industry sector")
    executive_summary: str = Field(description="2-3 sentence executive summary")
    ml_findings: List[MLFinding] = Field(description="ML-backed analytical findings")
    ai_opportunities: List[AIOpportunity] = Field(description="Recommended AI interventions")
    roi_projection: ROIProjection = Field(description="Financial impact projection")
    ethical_considerations: str = Field(description="Ethical risks and mitigations")
    risk_assessment: str = Field(description="Implementation risks and contingencies")

What does this code do?

Each class defines a piece of the final executive proposal. MLFinding captures individual ML insights with confidence scores. AIOpportunity requires every recommendation to include an ethical risk/benefit statement.

ROIProjection forces the agent to produce concrete financial numbers. BusinessProposal ties everything together into one validated structure. When CrewAI uses output_pydantic=BusinessProposal on a task, the agent must produce output matching this exact schema or the task fails.

  • Save models.py.
  • Verify the models parse correctly by running:
python -c "from models import BusinessProposal; print(f'BusinessProposal has {len(BusinessProposal.model_fields)} fields')"

You should see BusinessProposal has 8 fields printed in your terminal.

Getting a validation error?

Check that your models.py file starts with from pydantic import BaseModel, Field and from typing import List. Also ensure every Field() has a description keyword argument.

help me fix my Pydantic models in models.py

✔️ Awesome, I've got everything!

Great. Double check you've saved models.py with all four Pydantic classes.

ⓧ I'd like to double check the full code

from pydantic import BaseModel, Field
from typing import List


class MLFinding(BaseModel):
    technique: str = Field(description="ML technique used")
    insight: str = Field(description="Key finding from the analysis")
    confidence: float = Field(description="Confidence score 0-1")
    evidence: str = Field(description="Numeric evidence supporting the finding")


class AIOpportunity(BaseModel):
    area: str = Field(description="Business area for AI intervention")
    solution: str = Field(description="Proposed GenAI/agentic solution")
    expected_impact: str = Field(description="Quantified expected impact")
    risk_benefit: str = Field(description="Ethical risk/benefit statement")


class ROIProjection(BaseModel):
    cost_savings_annual: float = Field(description="Projected annual cost savings in USD")
    revenue_uplift_percent: float = Field(description="Projected revenue increase percentage")
    customer_satisfaction_delta: float = Field(description="Expected CSAT improvement points")
    payback_period_months: int = Field(description="Months to positive ROI")


class BusinessProposal(BaseModel):
    company_name: str = Field(description="Target company name")
    industry: str = Field(description="Industry sector")
    executive_summary: str = Field(description="2-3 sentence executive summary")
    ml_findings: List[MLFinding] = Field(description="ML-backed analytical findings")
    ai_opportunities: List[AIOpportunity] = Field(description="Recommended AI interventions")
    roi_projection: ROIProjection = Field(description="Financial impact projection")
    ethical_considerations: str = Field(description="Ethical risks and mitigations")
    risk_assessment: str = Field(description="Implementation risks and contingencies")

Your ML toolkit is complete. Next up, you'll define the four AI agents and wire these tools into their workflows with task guardrails that enforce output quality.

Define the Agent Crew with Guardrails

Your ML toolkit is ready. Now you need the team that will use it.

In a real data science consulting engagement, raw model outputs are never sent to executives. They pass through layers of validation, context-building, and human review. In this step, you will define four AI agents that mirror a professional consulting team, wire quality guardrails onto their tasks, and gate the final deliverable on your personal approval.

In this step, get ready to:
  • Define four specialized agents (Data Engineer, ML Analyst, AI Strategist, Executive Advisor).
  • Create tasks with guardrails that enforce output quality, numeric evidence, and ethical transparency.
  • Implement guardrail functions that validate agent outputs and trigger retries on failure.
Define the four AI agents

Each agent in a CrewAI crew has a role, goal, and backstory that shapes how the LLM behaves. Think of it like a job description that tells the model exactly what perspective to bring.

  • Create a new file called crew.py in your project folder.
  • Add the following imports and environment setup at the top:
import os
import tiktoken
from dotenv import load_dotenv

import mlflow
from crewai import Agent, Task, Crew, Process
from crewai.hooks import before_llm_call, after_llm_call, LLMCallHookContext

from models import BusinessProposal
from ml_tools import (
    detect_anomalies,
    segment_customers,
    predict_churn_risk,
    check_ethical_bias,
)
from guardrails import (
    validate_data_completeness,
    validate_numeric_evidence,
    validate_ethical_transparency,
    validate_proposal_structure,
)

load_dotenv()

What do these imports do?

  • Agent, Task, Crew, and Process are the core CrewAI building blocks for defining your multi-agent system.
  • The ml_tools and guardrails imports bring in the custom tools and validation functions you will build in this step.
  • tiktoken, mlflow, and the hook imports will be used in the next step for cost tracking and observability.
  • Below the imports, add the four agent definitions:
# --- Agent Definitions ---
data_engineer = Agent(
    role="Senior Data Engineer",
    goal="Validate, preprocess, and profile incoming business data to ensure "
         "downstream ML analysis has clean, complete inputs",
    backstory=(
        "You are a meticulous data engineer with 8 years of experience in "
        "enterprise data pipelines. You catch data quality issues that others miss "
        "and produce clear data profiling reports that help analysts work faster."
    ),
    llm="gpt-4o-mini",
    verbose=True,
    allow_delegation=False,
    max_iter=10,
)

ml_analyst = Agent(
    role="Machine Learning Analyst",
    goal="Apply statistical and ML techniques to extract actionable patterns "
         "from business data, providing evidence-backed insights",
    backstory=(
        "You are a data scientist specializing in applied ML for business. "
        "Your strength is translating complex model outputs into clear, "
        "actionable insights that non-technical stakeholders can act on. "
        "You always cite specific numbers from your analysis."
    ),
    llm="gpt-4o-mini",
    verbose=True,
    allow_delegation=False,
    tools=[detect_anomalies, segment_customers, predict_churn_risk],
    max_iter=15,
)

ai_strategist = Agent(
    role="AI Strategy Consultant",
    goal="Map ML findings to specific Generative AI and agentic automation "
         "opportunities that drive measurable business outcomes",
    backstory=(
        "You are a senior AI consultant who has led transformation projects "
        "across healthcare, finance, manufacturing, and retail. You connect "
        "data insights to concrete AI solutions with realistic ROI projections. "
        "You always consider ethical implications and include risk/benefit "
        "statements for every recommendation."
    ),
    llm="gpt-4o-mini",
    verbose=True,
    allow_delegation=False,
    tools=[check_ethical_bias],
    max_iter=10,
)

executive_advisor = Agent(
    role="Executive Business Advisor",
    goal="Synthesize all analysis into a compelling, structured executive "
         "proposal that demonstrates clear ROI and addresses ethical governance",
    backstory=(
        "You are a C-suite advisor who translates technical AI capabilities "
        "into business language. Your proposals have secured over $50M in AI "
        "investments. You produce structured, data-driven recommendations "
        "that satisfy both CFOs and Chief Ethics Officers."
    ),
    llm="gpt-4o-mini",
    verbose=True,
    allow_delegation=False,
    max_iter=10,
)

How do agent roles shape LLM behavior?

The role, goal, and backstory form a system prompt that constrains how gpt-4o-mini responds. A "meticulous data engineer" produces different output than a "C-suite advisor" given the same data.

The tools parameter gives agents access to specific functions. Only the ML Analyst can run the ML tools, and only the AI Strategist can run the ethical bias checker. This mirrors real teams where each specialist owns their domain.

  • Save crew.py.

You will not be able to run this file yet because the guardrails module does not exist. You will create it in substep 3.

Seeing a red underline on the guardrails import?

That is expected. Your editor flags the import because guardrails.py does not exist yet. The warning will disappear once you create the file later in this step.

help me understand why my import is showing an error in VS Code

Define tasks with guardrails and context chaining

Tasks tell each agent what to do and how to validate the result. A guardrail is a Python function that inspects the agent's output and either approves it or sends it back for a retry with feedback.

  • In crew.py, add the first two task definitions below your agent definitions:
# --- Task Definitions ---
data_engineering_task = Task(
    description=(
        "Profile and validate the business data for {company_name} in the "
        "{industry} industry. Their pain points are: {pain_points}. "
        "Current KPIs: {current_kpis}. "
        "Assess data completeness, identify data quality issues, and produce "
        "a summary of available metrics for downstream ML analysis."
    ),
    expected_output=(
        "A data profiling report covering: available metrics (revenue, customers, "
        "churn, satisfaction, costs), data quality assessment, and readiness "
        "score for ML analysis."
    ),
    agent=data_engineer,
    guardrail=validate_data_completeness,
    guardrail_max_retries=2,
)

ml_analysis_task = Task(
    description=(
        "Using the validated data profile, run ML analysis on {company_name}'s "
        "business metrics. Execute anomaly detection on revenue/operations, "
        "customer segmentation, and churn risk prediction. "
        "Report findings with specific numbers and confidence scores."
    ),
    expected_output=(
        "ML analysis report with: anomaly detection results (which months, "
        "severity), customer segments (sizes, characteristics), churn risk "
        "distribution (high/medium/low counts), and key predictive factors."
    ),
    agent=ml_analyst,
    guardrails=[validate_numeric_evidence],
    guardrail_max_retries=2,
    context=[data_engineering_task],
)

What do these task parameters do?

  • guardrail (singular) accepts one validation function. guardrails (plural) accepts a list of functions executed sequentially.
  • guardrail_max_retries=2 means if validation fails, the agent retries the task up to 2 times with the error message as feedback.
  • context=[data_engineering_task] passes the Data Engineer's output into the ML Analyst's prompt, creating a handoff chain.
  • The {company_name} placeholders are filled at runtime when you call crew.kickoff(inputs={...}).
  • Add the final two task definitions below the ones you just added:
ai_strategy_task = Task(
    description=(
        "Based on the ML findings, propose 3-5 specific Generative AI and "
        "agentic automation opportunities for {company_name} ({industry}). "
        "Each opportunity must: 1) directly address an ML-identified pattern, "
        "2) include a quantified expected impact, 3) include an ethical "
        "risk/benefit statement. Run the ethical bias checker on your "
        "recommendations before finalizing."
    ),
    expected_output=(
        "A strategic recommendations document with 3-5 AI opportunities, each "
        "containing: the business area, proposed solution, expected impact with "
        "numbers, and an ethical risk/benefit statement."
    ),
    agent=ai_strategist,
    guardrails=[validate_ethical_transparency],
    guardrail_max_retries=2,
    context=[ml_analysis_task],
)

executive_proposal_task = Task(
    description=(
        "Compile all findings into a structured executive proposal for "
        "{company_name}. Include: executive summary, ML-backed findings, "
        "AI opportunities with ROI projections, ethical considerations, "
        "and risk assessment. This will be reviewed by the Chief Data Officer "
        "before delivery. Make sure to check with the human reviewer if the "
        "proposal meets their standards."
    ),
    expected_output=(
        "A complete business proposal with executive summary, ML findings, "
        "AI opportunities, ROI projection (cost savings, revenue uplift, "
        "CSAT improvement, payback period), ethical considerations, and "
        "risk assessment."
    ),
    agent=executive_advisor,
    guardrails=[validate_proposal_structure],
    guardrail_max_retries=2,
    human_input=True,
    output_pydantic=BusinessProposal,
    context=[ml_analysis_task, ai_strategy_task],
)

What makes the executive task special?

human_input=True pauses execution and prompts you in the terminal to review the draft before the agent finalizes. You act as the Chief Data Officer approving the proposal.

output_pydantic=BusinessProposal forces the agent to structure its response into the exact Pydantic model you defined in models.py. If the output does not match the schema, CrewAI retries automatically.

context=[ml_analysis_task, ai_strategy_task] feeds both upstream outputs into this final task, giving the Executive Advisor the full picture.

  • Save crew.py.

Seeing a NameError or undefined variable warning?

Make sure validate_data_completeness, validate_numeric_evidence, validate_ethical_transparency, and validate_proposal_structure are all listed in the from guardrails import (...) block at the top. You will create the actual functions next.

help me fix import errors in my crew.py file

Implement the guardrail validation functions

Guardrails are plain Python functions that accept a TaskOutput and return a tuple: (True, result) if the output passes, or (False, feedback_message) if it needs a retry. The feedback message tells the agent exactly what to fix.

  • Create a new file called guardrails.py in your project folder.
  • Add the first two guardrail functions by pasting this code:
from typing import Tuple, Any
from crewai import TaskOutput


def validate_data_completeness(result: TaskOutput) -> Tuple[bool, Any]:
    """Ensure data analysis contains required metrics."""
    required_terms = ["revenue", "customer", "churn"]
    content_lower = result.raw.lower()
    missing = [term for term in required_terms if term not in content_lower]
    if missing:
        return (False, f"Missing required data points: {', '.join(missing)}. "
                       f"Please include analysis of all business metrics.")
    return (True, result.raw)


def validate_numeric_evidence(result: TaskOutput) -> Tuple[bool, Any]:
    """Ensure ML findings contain numeric evidence."""
    has_numbers = any(char.isdigit() for char in result.raw)
    has_percent = "%" in result.raw
    has_dollar = "$" in result.raw
    if not has_numbers:
        return (False, "Output lacks numeric evidence. Include specific numbers, percentages, or dollar amounts from ML analysis.")
    if not (has_percent or has_dollar):
        return (False, "Output needs quantified metrics. Include at least one percentage or dollar figure to support findings.")
    return (True, result.raw)

How do guardrails trigger retries?

When a guardrail returns (False, message), CrewAI sends the message back to the agent as feedback. The agent then retries the task, incorporating the feedback. This loop runs up to guardrail_max_retries times.

validate_data_completeness checks that the Data Engineer mentions revenue, customer, and churn metrics. validate_numeric_evidence ensures the ML Analyst includes actual numbers (not just prose) in the analysis.

  • Add the remaining two guardrail functions below the ones you just pasted:
def validate_ethical_transparency(result: TaskOutput) -> Tuple[bool, Any]:
    """Ensure every recommendation includes ethical risk/benefit."""
    content_lower = result.raw.lower()
    ethical_terms = ["risk", "ethical", "bias", "fair", "transparent", "accountab", "responsible"]
    has_ethical = any(term in content_lower for term in ethical_terms)
    if not has_ethical:
        return (False, "Recommendations must include ethical considerations. Add risk/benefit statements addressing fairness, transparency, or accountability for each suggestion.")
    return (True, result.raw)


def validate_proposal_structure(result: TaskOutput) -> Tuple[bool, Any]:
    """Ensure final proposal meets minimum quality standards."""
    word_count = len(result.raw.split())
    if word_count < 200:
        return (False, f"Proposal too brief ({word_count} words). Expand to at least 200 words with detailed analysis.")
    required_sections = ["summary", "finding", "opportunity", "roi"]
    content_lower = result.raw.lower()
    missing = [s for s in required_sections if s not in content_lower]
    if missing:
        return (False, f"Proposal missing sections: {', '.join(missing)}. Include executive summary, ML findings, AI opportunities, and ROI.")
    return (True, result.raw)

What do the ethical and structural guardrails enforce?

validate_ethical_transparency scans the AI Strategist's output for ethical language. If no risk/benefit discussion exists, the agent must redo the recommendation with ethical considerations included.

validate_proposal_structure enforces two things on the Executive Advisor's output: a minimum 200-word length and the presence of required sections (summary, findings, opportunities, ROI). This prevents thin proposals from reaching the human reviewer.

  • Save guardrails.py.
  • Switch back to crew.py and confirm the red underline on the guardrails import has disappeared. Your editor now recognizes the module.

Still seeing an import error in crew.py?

Make sure guardrails.py is in the same folder as crew.py. Check that the function names match exactly: validate_data_completeness, validate_numeric_evidence, validate_ethical_transparency, validate_proposal_structure.

help me debug why my guardrails import is failing in crew.py

✔️ Awesome, I've got everything!

Great. Double check that you have saved both guardrails.py and crew.py.

ⓧ I'd like to double check the full code

Here is the complete guardrails.py file:

from typing import Tuple, Any
from crewai import TaskOutput


def validate_data_completeness(result: TaskOutput) -> Tuple[bool, Any]:
    """Ensure data analysis contains required metrics."""
    required_terms = ["revenue", "customer", "churn"]
    content_lower = result.raw.lower()
    missing = [term for term in required_terms if term not in content_lower]
    if missing:
        return (False, f"Missing required data points: {', '.join(missing)}. "
                       f"Please include analysis of all business metrics.")
    return (True, result.raw)


def validate_numeric_evidence(result: TaskOutput) -> Tuple[bool, Any]:
    """Ensure ML findings contain numeric evidence."""
    has_numbers = any(char.isdigit() for char in result.raw)
    has_percent = "%" in result.raw
    has_dollar = "$" in result.raw
    if not has_numbers:
        return (False, "Output lacks numeric evidence. Include specific numbers, percentages, or dollar amounts from ML analysis.")
    if not (has_percent or has_dollar):
        return (False, "Output needs quantified metrics. Include at least one percentage or dollar figure to support findings.")
    return (True, result.raw)


def validate_ethical_transparency(result: TaskOutput) -> Tuple[bool, Any]:
    """Ensure every recommendation includes ethical risk/benefit."""
    content_lower = result.raw.lower()
    ethical_terms = ["risk", "ethical", "bias", "fair", "transparent", "accountab", "responsible"]
    has_ethical = any(term in content_lower for term in ethical_terms)
    if not has_ethical:
        return (False, "Recommendations must include ethical considerations. Add risk/benefit statements addressing fairness, transparency, or accountability for each suggestion.")
    return (True, result.raw)


def validate_proposal_structure(result: TaskOutput) -> Tuple[bool, Any]:
    """Ensure final proposal meets minimum quality standards."""
    word_count = len(result.raw.split())
    if word_count < 200:
        return (False, f"Proposal too brief ({word_count} words). Expand to at least 200 words with detailed analysis.")
    required_sections = ["summary", "finding", "opportunity", "roi"]
    content_lower = result.raw.lower()
    missing = [s for s in required_sections if s not in content_lower]
    if missing:
        return (False, f"Proposal missing sections: {', '.join(missing)}. Include executive summary, ML findings, AI opportunities, and ROI.")
    return (True, result.raw)

Your crew is defined and guarded. In the next step, you will add observability hooks that track every token and dollar spent as the agents collaborate.

Add Observability, Cost Tracking, and Tracing

Your agents, tools, and guardrails are ready. But in a production system, you need to know what your agents are doing, how much they cost, and where things go wrong.

Without observability, a multi-agent pipeline is a black box. In this step, you will wire up MLflow tracing, real-time token cost tracking, and post-execution metrics so you can see exactly what happens inside the crew.

In this step, get ready to:
  • Enable MLflow autologging and configure the Crew with tracing and memory.
  • Implement LLM hooks that track token usage and estimate cost in real time.
  • Build the run function that displays post-execution metrics and saves the final proposal.
Enable MLflow autologging and assemble the Crew

MLflow's CrewAI integration automatically captures every agent decision, tool call, and task handoff as a trace. Two lines of setup give you a full visual timeline in the MLflow dashboard you started earlier.

  • In crew.py, find the import section at the top of the file. Add these new imports alongside the existing ones:
import tiktoken

import mlflow
from crewai.hooks import before_llm_call, after_llm_call, LLMCallHookContext
  • Directly below the load_dotenv() line, add the MLflow configuration:
# --- MLflow Setup ---
mlflow.crewai.autolog()
mlflow.set_experiment("AI-Business-Intelligence-Crew")

What does this code do?

These two imports and two config lines set up full observability for your crew.

  • tiktoken is OpenAI's tokenizer library. You will use it to count tokens before and after each LLM call.
  • mlflow.crewai.autolog() automatically traces every agent step, tool invocation, and task completion into MLflow.
  • mlflow.set_experiment("AI-Business-Intelligence-Crew") groups all traces under a named experiment so they are easy to find in the dashboard.

Now you need the Crew object itself. The Crew ties your four agents and four tasks together into one sequential pipeline.

  • Scroll to the bottom of crew.py, below your task definitions. Add the Crew assembly:
# --- Crew Assembly ---
business_intelligence_crew = Crew(
    agents=[data_engineer, ml_analyst, ai_strategist, executive_advisor],
    tasks=[
        data_engineering_task,
        ml_analysis_task,
        ai_strategy_task,
        executive_proposal_task,
    ],
    process=Process.sequential,
    verbose=True,
    memory=True,
    planning=True,
    tracing=True,
)

What does this code do?

  • process=Process.sequential runs agents one after another, each building on the previous output.
  • memory=True enables shared memory so later agents can recall earlier findings.
  • planning=True tells CrewAI to generate an execution plan before running tasks.
  • tracing=True enables CrewAI's built-in tracing layer alongside MLflow's autologging.
  • Save crew.py.

Seeing a NameError on the Crew line?

Make sure Crew is included in your from crewai import line at the top of the file. It should read from crewai import Agent, Task, Crew, Process.

Help me fix a NameError in crew.py

Implement LLM hooks for real-time cost tracking

Every time an agent calls the LLM, you want to know how many tokens it used and what that costs. CrewAI's hook system lets you intercept every LLM call globally without modifying individual agents.

  • In crew.py, add the cost tracking state and the input hook directly below the mlflow.set_experiment(...) line (above your agent definitions):
# --- Cost Tracking State ---
token_usage = {"input_tokens": 0, "output_tokens": 0}

INPUT_COST_PER_MILLION = 0.15  # gpt-4o-mini input
OUTPUT_COST_PER_MILLION = 0.60  # gpt-4o-mini output


@before_llm_call
def track_input_tokens(context: LLMCallHookContext) -> None:
    """Count input tokens before each LLM call."""
    encoding = tiktoken.get_encoding("cl100k_base")
    total = sum(
        len(encoding.encode(msg.get("content", "")))
        for msg in context.messages
    )
    token_usage["input_tokens"] += total
    print(f"  [Cost] Input tokens this call: ~{total}")
    return None

What does this code do?

  • token_usage is a module-level dictionary that accumulates token counts across all LLM calls during the run.
  • The two cost constants match gpt-4o-mini pricing: $0.15 per million input tokens and $0.60 per million output tokens.
  • @before_llm_call fires before every LLM request across all agents. It uses tiktoken's cl100k_base encoding to count tokens in the message payload.
  • Directly below the track_input_tokens function, add the output hook:
@after_llm_call
def track_output_tokens(context: LLMCallHookContext) -> None:
    """Count output tokens and print running cost after each LLM call."""
    if context.response:
        encoding = tiktoken.get_encoding("cl100k_base")
        tokens = len(encoding.encode(context.response))
        token_usage["output_tokens"] += tokens
        cost = (
            (token_usage["input_tokens"] / 1_000_000) * INPUT_COST_PER_MILLION
            + (token_usage["output_tokens"] / 1_000_000) * OUTPUT_COST_PER_MILLION
        )
        print(f"  [Cost] Running total: ${cost:.4f}")
    return None

What does this code do?

@after_llm_call fires after each LLM response arrives. It encodes the response to count output tokens, then calculates the cumulative cost across the entire run. Every agent call prints a running dollar total so you can watch spend in real time.

  • Save crew.py.

You will not see a change yet. The hooks activate when the crew runs in the next substep.

Not sure where to place the hooks?

The hooks and cost constants must sit between the mlflow.set_experiment(...) line and your first Agent(...) definition. They are module-level code that registers globally before any agent is created.

Help me figure out where to place the LLM hooks in crew.py

Build the execution function with post-run metrics

The final piece is the function that actually kicks off the crew, feeds it a real business scenario, and prints a cost and metrics summary when it finishes.

  • At the very bottom of crew.py, below the business_intelligence_crew definition, add the run function:
def run_crew():
    """Execute the Business Intelligence Crew."""
    print("\n" + "=" * 60)
    print("  AI BUSINESS INTELLIGENCE CREW")
    print("  Production-Grade Multi-Agent System")
    print("=" * 60 + "\n")

    inputs = {
        "company_name": "MedFlow Health Systems",
        "industry": "Healthcare",
        "pain_points": (
            "High patient churn, inconsistent appointment scheduling, "
            "rising operational costs, low digital engagement"
        ),
        "current_kpis": (
            "Monthly revenue: $12M, Active patients: 35000, "
            "Avg satisfaction: 3.8/5, Monthly churn: 8%, "
            "Operational cost ratio: 72%"
        ),
    }

    print(f"Analyzing: {inputs['company_name']} ({inputs['industry']})")
    print(f"Pain Points: {inputs['pain_points']}")
    print("-" * 60 + "\n")

    result = business_intelligence_crew.kickoff(inputs=inputs)

    # --- Post-Execution Report ---
    print("\n" + "=" * 60)
    print("  EXECUTION COMPLETE")
    print("=" * 60)

    # Cost summary
    total_cost = (
        (token_usage["input_tokens"] / 1_000_000) * INPUT_COST_PER_MILLION
        + (token_usage["output_tokens"] / 1_000_000) * OUTPUT_COST_PER_MILLION
    )
    print(f"\n  Token Usage:")
    print(f"    Input tokens:  {token_usage['input_tokens']:,}")
    print(f"    Output tokens: {token_usage['output_tokens']:,}")
    print(f"    Total cost:    ${total_cost:.4f}")

    # Usage metrics from CrewAI
    print(f"\n  CrewAI Usage Metrics:")
    print(f"    {business_intelligence_crew.usage_metrics}")

    # Save output
    output_path = "output/proposal.md"
    os.makedirs("output", exist_ok=True)
    with open(output_path, "w") as f:
        f.write(f"# AI Transformation Proposal: {inputs['company_name']}\n\n")
        f.write(result.raw)
    print(f"\n  Proposal saved to: {output_path}")
    print(f"  MLflow dashboard: http://127.0.0.1:5000")
    print("=" * 60)

    return result


if __name__ == "__main__":
    run_crew()

What does this code do?

  • The inputs dictionary defines a healthcare company scenario. These values get injected into each task's {company_name}, {industry}, {pain_points}, and {current_kpis} placeholders.
  • business_intelligence_crew.kickoff(inputs=inputs) launches the sequential pipeline. Each agent runs its task in order.
  • After execution, the function prints your custom token tracking totals alongside CrewAI's own usage_metrics (total_tokens, prompt_tokens, completion_tokens, successful_requests).
  • The final proposal is saved to output/proposal.md so you have a persistent record of the crew's work.
  • Save crew.py.

Why two sources of token metrics?

Your custom hooks give you real-time per-call visibility as the crew runs. CrewAI's usage_metrics provides the official totals after execution. Comparing them helps you verify your tracking is accurate and catch discrepancies from retries or guardrail re-runs.

Seeing an IndentationError or SyntaxError?

Make sure the run_crew() function is at the top level (no extra indentation). The if __name__ == "__main__": block should be the very last thing in the file.

Help me fix a syntax error in my run_crew function

✔️ Awesome, I've got everything!

Great. Double check you have saved crew.py and that there are no red underlines in your editor.

ⓧ I'd like to double check the full code

Here is the complete crew.py file. Compare your version against this:

import os
import tiktoken
from dotenv import load_dotenv

import mlflow
from crewai import Agent, Task, Crew, Process
from crewai.hooks import before_llm_call, after_llm_call, LLMCallHookContext

from models import BusinessProposal
from ml_tools import detect_anomalies, segment_customers, predict_churn_risk, check_ethical_bias
from guardrails import validate_data_completeness, validate_numeric_evidence, validate_ethical_transparency, validate_proposal_structure

load_dotenv()

mlflow.crewai.autolog()
mlflow.set_experiment("AI-Business-Intelligence-Crew")

token_usage = {"input_tokens": 0, "output_tokens": 0}
INPUT_COST_PER_MILLION = 0.15
OUTPUT_COST_PER_MILLION = 0.60

@before_llm_call
def track_input_tokens(context: LLMCallHookContext) -> None:
    encoding = tiktoken.get_encoding("cl100k_base")
    total = sum(len(encoding.encode(msg.get("content", ""))) for msg in context.messages)
    token_usage["input_tokens"] += total
    print(f"  [Cost] Input tokens this call: ~{total}")
    return None

@after_llm_call
def track_output_tokens(context: LLMCallHookContext) -> None:
    if context.response:
        encoding = tiktoken.get_encoding("cl100k_base")
        tokens = len(encoding.encode(context.response))
        token_usage["output_tokens"] += tokens
        cost = ((token_usage["input_tokens"] / 1_000_000) * INPUT_COST_PER_MILLION + (token_usage["output_tokens"] / 1_000_000) * OUTPUT_COST_PER_MILLION)
        print(f"  [Cost] Running total: ${cost:.4f}")
    return None

data_engineer = Agent(role="Senior Data Engineer", goal="Validate, preprocess, and profile incoming business data to ensure downstream ML analysis has clean, complete inputs", backstory="You are a meticulous data engineer with 8 years of experience in enterprise data pipelines. You catch data quality issues that others miss and produce clear data profiling reports that help analysts work faster.", llm="gpt-4o-mini", verbose=True, allow_delegation=False, max_iter=10)
ml_analyst = Agent(role="Machine Learning Analyst", goal="Apply statistical and ML techniques to extract actionable patterns from business data, providing evidence-backed insights", backstory="You are a data scientist specializing in applied ML for business. Your strength is translating complex model outputs into clear, actionable insights that non-technical stakeholders can act on. You always cite specific numbers from your analysis.", llm="gpt-4o-mini", verbose=True, allow_delegation=False, tools=[detect_anomalies, segment_customers, predict_churn_risk], max_iter=15)
ai_strategist = Agent(role="AI Strategy Consultant", goal="Map ML findings to specific Generative AI and agentic automation opportunities that drive measurable business outcomes", backstory="You are a senior AI consultant who has led transformation projects across healthcare, finance, manufacturing, and retail. You connect data insights to concrete AI solutions with realistic ROI projections. You always consider ethical implications and include risk/benefit statements for every recommendation.", llm="gpt-4o-mini", verbose=True, allow_delegation=False, tools=[check_ethical_bias], max_iter=10)
executive_advisor = Agent(role="Executive Business Advisor", goal="Synthesize all analysis into a compelling, structured executive proposal that demonstrates clear ROI and addresses ethical governance", backstory="You are a C-suite advisor who translates technical AI capabilities into business language. Your proposals have secured over $50M in AI investments. You produce structured, data-driven recommendations that satisfy both CFOs and Chief Ethics Officers.", llm="gpt-4o-mini", verbose=True, allow_delegation=False, max_iter=10)

data_engineering_task = Task(description="Profile and validate the business data for {company_name} in the {industry} industry. Their pain points are: {pain_points}. Current KPIs: {current_kpis}. Assess data completeness, identify data quality issues, and produce a summary of available metrics for downstream ML analysis.", expected_output="A data profiling report covering: available metrics (revenue, customers, churn, satisfaction, costs), data quality assessment, and readiness score for ML analysis.", agent=data_engineer, guardrail=validate_data_completeness, guardrail_max_retries=2)
ml_analysis_task = Task(description="Using the validated data profile, run ML analysis on {company_name}'s business metrics. Execute anomaly detection on revenue/operations, customer segmentation, and churn risk prediction. Report findings with specific numbers and confidence scores.", expected_output="ML analysis report with: anomaly detection results (which months, severity), customer segments (sizes, characteristics), churn risk distribution (high/medium/low counts), and key predictive factors.", agent=ml_analyst, guardrails=[validate_numeric_evidence], guardrail_max_retries=2, context=[data_engineering_task])
ai_strategy_task = Task(description="Based on the ML findings, propose 3-5 specific Generative AI and agentic automation opportunities for {company_name} ({industry}). Each opportunity must: 1) directly address an ML-identified pattern, 2) include a quantified expected impact, 3) include an ethical risk/benefit statement. Run the ethical bias checker on your recommendations before finalizing.", expected_output="A strategic recommendations document with 3-5 AI opportunities, each containing: the business area, proposed solution, expected impact with numbers, and an ethical risk/benefit statement.", agent=ai_strategist, guardrails=[validate_ethical_transparency], guardrail_max_retries=2, context=[ml_analysis_task])
executive_proposal_task = Task(description="Compile all findings into a structured executive proposal for {company_name}. Include: executive summary, ML-backed findings, AI opportunities with ROI projections, ethical considerations, and risk assessment. This will be reviewed by the Chief Data Officer before delivery. Make sure to check with the human reviewer if the proposal meets their standards.", expected_output="A complete business proposal with executive summary, ML findings, AI opportunities, ROI projection (cost savings, revenue uplift, CSAT improvement, payback period), ethical considerations, and risk assessment.", agent=executive_advisor, guardrails=[validate_proposal_structure], guardrail_max_retries=2, human_input=True, output_pydantic=BusinessProposal, context=[ml_analysis_task, ai_strategy_task])

business_intelligence_crew = Crew(agents=[data_engineer, ml_analyst, ai_strategist, executive_advisor], tasks=[data_engineering_task, ml_analysis_task, ai_strategy_task, executive_proposal_task], process=Process.sequential, verbose=True, memory=True, planning=True, tracing=True)

def run_crew():
    print("\n" + "=" * 60)
    print("  AI BUSINESS INTELLIGENCE CREW")
    print("  Production-Grade Multi-Agent System")
    print("=" * 60 + "\n")
    inputs = {"company_name": "MedFlow Health Systems", "industry": "Healthcare", "pain_points": "High patient churn, inconsistent appointment scheduling, rising operational costs, low digital engagement", "current_kpis": "Monthly revenue: $12M, Active patients: 35000, Avg satisfaction: 3.8/5, Monthly churn: 8%, Operational cost ratio: 72%"}
    print(f"Analyzing: {inputs['company_name']} ({inputs['industry']})")
    print(f"Pain Points: {inputs['pain_points']}")
    print("-" * 60 + "\n")
    result = business_intelligence_crew.kickoff(inputs=inputs)
    print("\n" + "=" * 60)
    print("  EXECUTION COMPLETE")
    print("=" * 60)
    total_cost = ((token_usage["input_tokens"] / 1_000_000) * INPUT_COST_PER_MILLION + (token_usage["output_tokens"] / 1_000_000) * OUTPUT_COST_PER_MILLION)
    print(f"\n  Token Usage:")
    print(f"    Input tokens:  {token_usage['input_tokens']:,}")
    print(f"    Output tokens: {token_usage['output_tokens']:,}")
    print(f"    Total cost:    ${total_cost:.4f}")
    print(f"\n  CrewAI Usage Metrics:")
    print(f"    {business_intelligence_crew.usage_metrics}")
    output_path = "output/proposal.md"
    os.makedirs("output", exist_ok=True)
    with open(output_path, "w") as f:
        f.write(f"# AI Transformation Proposal: {inputs['company_name']}\n\n")
        f.write(result.raw)
    print(f"\n  Proposal saved to: {output_path}")
    print(f"  MLflow dashboard: http://127.0.0.1:5000")
    print("=" * 60)
    return result

if __name__ == "__main__":
    run_crew()

Your crew is fully wired with observability, cost tracking, and tracing. Next up, you will run the full pipeline and interact with the human-in-the-loop approval gate.

Run the Crew with Human-in-the-Loop

Your agents are built, your tools are wired, your guardrails are in place, and MLflow is tracking everything. Now it is time to see the full system in action.

In this step, you will execute the entire crew pipeline and play the role of Chief Data Officer. When the Executive Advisor finishes drafting its proposal, the pipeline pauses and asks for your approval before delivering the final output.

In this step, get ready to:
  • Launch the multi-agent crew and observe the sequential execution.
  • Review and approve the executive proposal through the human-in-the-loop prompt.
  • Explore the MLflow trace and review the final structured output.
Launch the crew pipeline

The run_crew() function in crew.py assembles all four agents and their tasks into a sequential Crew with planning=True, memory=True, and tracing=True. It then calls crew.kickoff() with MedFlow Health Systems as the target company.

  • Run the crew by executing this command in your terminal:
python crew.py

What happens when you run this?

CrewAI first runs a planning phase where it maps out the execution strategy. Then each agent runs in sequence:

  • The Data Engineer validates and profiles the business data.
  • The ML Analyst calls all three ML tools (Isolation Forest, K-Means, Logistic Regression) and reports findings with numeric evidence.
  • The AI Strategist maps findings to GenAI opportunities and runs the ethical bias checker.
  • The Executive Advisor drafts the final proposal, then pauses for your human approval.

You will see verbose output in the terminal as each agent works. The LLM hooks print running cost totals after every API call.

This takes a few minutes

The full pipeline makes multiple LLM calls across four agents, plus planning. Expect 3-5 minutes of execution time depending on API latency. Watch the terminal for the [Cost] lines to track progress.

Seeing an API key error?

Make sure your .env file contains a valid OPENAI_API_KEY with an active OpenAI account that has billing enabled. The crew needs a funded API key to call gpt-4o-mini.

help me debug an OpenAI API key error in my CrewAI project

Approve the proposal as Chief Data Officer

When the pipeline reaches the Executive Advisor's task, CrewAI detects human_input=True and pauses execution. The agent presents its draft proposal in the terminal and asks for your input before finalizing.

  • Wait for the terminal to display the Executive Advisor's draft proposal followed by a prompt asking for your feedback.
  • Read through the draft proposal. It should contain an executive summary, ML findings, AI opportunities, ROI projections, and ethical considerations.
  • Type your feedback or approval into the terminal prompt, then press Enter.

What should I type?

You have two options. You can provide steering feedback like "Add more detail on compliance risks for healthcare data" or "Include specific HIPAA considerations." The agent will revise its proposal based on your input.

Alternatively, you can simply type "Approve" or "Looks good" to accept the draft as-is. The agent will then finalize and deliver its answer.

Why does human-in-the-loop matter?

In production AI systems, human-in-the-loop approval ensures that AI-generated recommendations pass through a human judgment gate before reaching stakeholders. This is how real enterprise teams maintain accountability.

Your feedback becomes part of the task context. The agent uses it to refine its output before delivering the structured BusinessProposal Pydantic object.

Pipeline seems stuck?

If the terminal appears frozen with no new output for more than 30 seconds, the agent is likely waiting for your input. Scroll up to check if a human input prompt appeared above the fold.

my CrewAI pipeline seems stuck and I can't find the human input prompt

Review the outputs and MLflow trace

After you approve the proposal, the pipeline completes. The terminal prints a token usage summary and saves the proposal to a file.

  • Check the terminal output for the token usage and cost summary.
  • Open the saved proposal by running:
cat output/proposal.md

You should see a formatted markdown file starting with # AI Transformation Proposal: MedFlow Health Systems followed by the full executive proposal.

  • Open your browser and navigate to http://127.0.0.1:5000 to view the MLflow dashboard.
  • Click on the AI-Business-Intelligence-Crew experiment in the left sidebar.
  • Click the most recent run to expand its trace.

What should I see in MLflow?

The trace view shows each agent's execution as a span. You can see which tools were called, how many tokens each LLM call consumed, the latency per task, and the total cost of the run.

Look for the ML Analyst's span in particular. It should show three tool calls: detect_anomalies, segment_customers, and predict_churn_risk. This level of observability is what separates production AI from prototype demos.

Your terminal also displays the CrewAI usage metrics including total tokens, prompt tokens, completion tokens, and successful requests.

MLflow dashboard is empty?

Make sure your MLflow server is still running in the separate terminal tab from earlier. If it was stopped, restart it with mlflow server and rerun python crew.py.

my MLflow dashboard shows no experiments after running my CrewAI crew

Your multi-agent AI system is fully operational. You have seen four agents collaborate, ML tools provide statistical evidence, guardrails enforce quality, cost tracking keep you informed, and human approval gate the final output. Next up, the Secret Mission challenges you to add an Ethics Auditor agent that scores every recommendation before delivery.

Secret mission

Add an AI Ethics Auditor Agent

Your crew produces solid proposals, but in production, every AI recommendation needs ethical vetting before it reaches a client. Add a 5th agent that scores every recommendation on fairness, transparency, and accountability using a custom scoring rubric. If any dimension falls below 70/100, the pipeline flags it for revision.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

This project runs entirely locally, so there are no ongoing cloud costs. The only cost was the OpenAI API calls you made during execution (approximately $0.30-$0.50 total).

Resources you used:

  • MLflow tracking server (running locally at http://127.0.0.1:5000).
  • Virtual environment with CrewAI, scikit-learn, MLflow, and dependencies.
  • Project files (crew.py, models.py, ml_tools.py, data_generator.py, guardrails.py, ethics_auditor.py, .env).
  • MLflow experiment data (mlruns/ directory) and generated proposal (output/proposal.md).

Keep everything running

No action needed. Choose this if you're still actively building or want to re-run the crew with different company scenarios.

  • Your MLflow server will keep running at http://127.0.0.1:5000 until you close the terminal or shut down your machine.
  • Your virtual environment stays active in the current terminal session.
  • All project files and experiment traces remain on your Desktop.

Pause - I'll come back to this later

Shut down running processes to free up system resources. Your files stay intact for next time.

  • Switch to the terminal where the MLflow server is running and press Ctrl+C to stop it.
  • In your project terminal, deactivate the virtual environment by running:
deactivate
  • Close VS Code.
  • To resume later, open the project folder in VS Code, reactivate your virtual environment, and restart the MLflow server with mlflow server.

Delete - I don't want to use this again

Remove all project files, the virtual environment, and MLflow data. This frees up disk space completely.

  • Switch to the terminal where the MLflow server is running and press Ctrl+C to stop it.
  • In your project terminal, deactivate the virtual environment by running:
deactivate
  • Close VS Code.
  • Navigate to your Desktop and delete the project folder. On macOS/Linux, run:
cd ~/Desktop
ls
  • Identify your project folder in the output, then delete it by running:
rm -rf [[PROJECT_FOLDER="your project folder name"]]
  • On Windows, open File Explorer, navigate to your Desktop, right-click the project folder, and select Delete.
  • Delete your .env file's contents from any password manager or notes app where you stored your OpenAI API key for this project.

Your OpenAI API key still exists

Deleting the project folder removes the local .env file, but your OpenAI API key itself still exists in your OpenAI account. If you want to revoke it, visit the API keys page in your OpenAI dashboard and delete the key there.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-grade multi-agent AI system that mirrors how real data science consulting teams deliver value to businesses.

You've learned how to:

  • Built a 4-agent data science crew using CrewAI that collaborates sequentially to produce AI transformation proposals backed by ML evidence.
  • Created custom ML tools (Isolation Forest anomaly detection, K-Means clustering, Logistic Regression churn scoring) exposed to agents via CrewAI's @tool decorator, providing statistical grounding for LLM recommendations.
  • Implemented production-grade guardrails, LLM hooks for real-time cost tracking, MLflow tracing for full observability, and human-in-the-loop approval to demonstrate AI governance in action.
  • (Secret Mission) Added an AI Ethics Auditor agent that gates every recommendation on fairness, transparency, and accountability scores before the final proposal is delivered.

Ready to quiz yourself?