Chapter 24 · Capstone 2: Data Analysis Pipeline

Chapter goals: Use Flow event-driven orchestration to build a "CSV dataset → cleaning advice → statistical analysis → report" pipeline, and master three production-grade techniques: Pydantic structured state, @router branching, and pandas wrapped as a tool.

24.1 Why This Pipeline Should Use Flow Instead of Crew

Unlike Chapter 23, data analysis has deterministic code steps (read CSV, compute statistics) and branching logic (small datasets analyzed directly, large datasets sampled first). Crews excel at role collaboration; Flows excel at precise control of execution paths. So the correct shape is:

@start load data (code) → @router judge size (code)
   ├─ small file ─────────────────→ analysis crew
   └─ large file → sample crew ──→ analysis crew
                                    ↓
                        @listen aggregate and generate report (crew)

Principle recap (Chapter 15): never use LLM for decisions that code can make deterministically; LLMs should only handle genuinely linguistic tasks like cleaning advice and conclusion interpretation.

24.2 Defining Flow State with Pydantic

Structured state gives each step's input and output type constraints — far more reliable than raw dicts:

# state.py —— Flow shared state definition
from pathlib import Path

from pydantic import BaseModel, Field


class DatasetInfo(BaseModel):
    """Step 1: metadata extracted from the dataset"""
    path: str
    rows: int
    cols: int
    columns: list[str] = Field(default_factory=list)
    missing_ratio: float = 0.0      # Whole-table missing rate, determines cleaning intensity
    size_class: str = "small"       # Written back by router: small / large


class CleaningAdvice(BaseModel):
    """Cleaning advice produced by the sampling crew (structured output)"""
    issues: list[str] = Field(description="Data quality issues found")
    recommendations: list[str] = Field(description="Cleaning recommendations, one per item")


class AnalysisState(BaseModel):
    """Root state for the entire Flow"""
    dataset: DatasetInfo | None = None
    cleaning: CleaningAdvice | None = None
    stats_summary: str = ""
    final_report: str = ""

    class Config:
        # Allow pandas and other objects to pass through state without validation blocking
        arbitrary_types_allowed = True


def probe_csv(path: str) -> DatasetInfo:
    """Pure-code step: probe CSV metadata without consuming any tokens"""
    import pandas as pd

    df = pd.read_csv(path)
    return DatasetInfo(
        path=str(Path(path).resolve()),
        rows=len(df),
        cols=len(df.columns),
        columns=list(df.columns),
        missing_ratio=float(df.isna().mean().mean()),
    )

probe_csv is a plain Python function — reads file, counts rows and columns, computes missing rate, zero LLM involvement. This is the value of Flow: separating "what can be computed" from "what needs to be asked of the model."

24.3 Wrapping pandas as a Tool

The analysis crew needs to genuinely "touch" the data. Wrap statistical computations as BaseTool so the model only decides which statistic to call and how to interpret the results:

# tools.py —— Data analysis toolset (practical application of Chapter 8 patterns)
import pandas as pd
from crewai.tools import BaseTool
from pydantic import BaseModel, Field


class DescribeInput(BaseModel):
    path: str = Field(description="Absolute path to CSV file")
    columns: list[str] = Field(default_factory=list, description="Only describe these columns; empty means all")


class DescribeDatasetTool(BaseTool):
    name: str = "describe_dataset"
    description: str = (
        "Compute count/mean/std/min/quartiles/max for CSV numeric columns. "
        "Use to understand field distributions and outliers."
    )
    args_schema: type[BaseModel] = DescribeInput

    def _run(self, path: str, columns: list[str] | None = None) -> str:
        df = pd.read_csv(path)
        if columns:
            df = df[columns]
        desc = df.describe(include="number").round(3)
        return desc.to_string()


class CorrelationInput(BaseModel):
    path: str = Field(description="Absolute path to CSV file")
    top_n: int = Field(default=5, description="Return the top N most correlated numeric column pairs")


class CorrelationTool(BaseTool):
    name: str = "correlation_matrix"
    description: str = "Compute Pearson correlation matrix of numeric columns, return top N correlated pairs and coefficients"
    args_schema: type[BaseModel] = CorrelationInput

    def _run(self, path: str, top_n: int = 5) -> str:
        corr = pd.read_csv(path).corr(numeric_only=True)
        pairs = (
            corr.where(lambda x: abs(x) < 0.999)          # Remove self-correlation 1.0
            .stack()
            .sort_values(key=abs, ascending=False)         # Sort by correlation strength
            .drop_duplicates()
        )
        head = pairs.head(top_n * 2).iloc[::2]             # Remove ±symmetry duplicates
        return "\n".join(f"{a} ~ {b}: {v:.3f}" for (a, b), v in head.items())

Key wrapping points (cross-reference Chapter 8): args_schema uses Pydantic for strict parameter validation — the model gets an immediate validation error if it fills parameters wrong; _run returns a compact string, not the full DataFrame — controlling token volume entering context is the lifeblood of data-type tools.

24.4 Main Flow: @start / @router / @listen Orchestration

# flow.py —— Data analysis pipeline main flow
import os

from crewai import Agent, Crew, Task, LLM
from crewai.flow.flow import Flow, listen, router, start
from pydantic import BaseModel

from state import AnalysisState, CleaningAdvice, probe_csv
from tools import CorrelationTool, DescribeDatasetTool

llm = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
)

ROW_THRESHOLD = 50_000   # Datasets over 50k rows are considered large


def make_cleaning_crew() -> Crew:
    advisor = Agent(
        role="Data Quality Advisor",
        goal="Review dataset overview and produce actionable cleaning advice",
        backstory="Senior data engineer who has seen every脏 data pattern; advice is always specific to column names and operations",
        llm=llm,
    )
    task = Task(
        description=(
            "Dataset {path} has {rows} rows and {cols} columns, overall missing rate "
            "{missing_ratio:.1%}. Assess quality risks and give cleaning advice."
        ),
        expected_output="Issue list and itemized recommendations (in Chinese)",
        agent=advisor,
        output_pydantic=CleaningAdvice,   # Structured output flows directly into state
    )
    return Crew(agents=[advisor], tasks=[task])


def make_analysis_crew() -> Crew:
    analyst = Agent(
        role="Data Analyst",
        goal="Complete exploratory analysis with statistical tools and interpret findings",
        backstory="Data scientist with 10 years experience, insists every conclusion must be backed by numbers",
        llm=llm,
        tools=[DescribeDatasetTool(), CorrelationTool()],
        max_iter=6,
    )
    task = Task(
        description=(
            "Perform exploratory analysis on {path}: first use describe_dataset to "
            "understand distributions, then use correlation_matrix to find strong pairs, "
            "finally provide 3 business interpretations."
        ),
        expected_output="Key statistics summary + correlation findings + 3 number-backed interpretations",
        agent=analyst,
    )
    return Crew(agents=[analyst], tasks=[task])


def make_report_crew() -> Crew:
    writer = Agent(
        role="Report Writer",
        goal="Integrate cleaning and analysis results into a management-readable report",
        backstory="Consultant skilled at translating technical details into business language",
        llm=llm,
    )
    task = Task(
        description="Write a report under 500 characters from the following materials:\nCleaning conclusions:\n{cleaning}\n\nAnalysis findings:\n{stats}",
        expected_output="Markdown report: background, data quality, key findings, action recommendations",
        agent=writer,
    )
    return Crew(agents=[writer], tasks=[task])


class DataPipeline(Flow[AnalysisState]):

    @start()
    def load_dataset(self):
        """Step 1: pure-code dataset probing, write to state"""
        self.state.dataset = probe_csv("data/sales.csv")
        print(f"[load] {self.state.dataset.rows} rows x {self.state.dataset.cols} columns")

    @router(load_dataset)
    def route_by_size(self):
        """Step 2: branch by row count — decision in code, not LLM"""
        if self.state.dataset.rows > ROW_THRESHOLD:
            return "large"
        return "small"

    @listen("large")
    def sample_and_advise(self):
        """Large file branch: sample first, then give cleaning advice"""
        sample_path = "data/_sample.csv"
        import pandas as pd
        pd.read_csv(self.state.dataset.path, nrows=10_000).to_csv(sample_path, index=False)
        advice = make_cleaning_crew().kickoff(
            inputs=self.state.dataset.model_dump() | {"path": sample_path}
        )
        self.state.cleaning = advice.pydantic

    @listen("small")
    def advise_directly(self):
        """Small file branch: give cleaning advice directly"""
        advice = make_cleaning_crew().kickoff(inputs=self.state.dataset.model_dump())
        self.state.cleaning = advice.pydantic

    @listen(sample_and_advise)   # In production, merge both branches with or_(...) — see note below
    def analyze(self):
        result = make_analysis_crew().kickoff(inputs={"path": self.state.dataset.path})
        self.state.stats_summary = result.raw

    @listen(analyze)
    def build_report(self):
        report = make_report_crew().kickoff(inputs={
            "cleaning": self.state.cleaning.model_dump_json(indent=2),
            "stats": self.state.stats_summary,
        })
        self.state.final_report = report.raw
        print("\n===== Final Report =====\n", self.state.final_report)


if __name__ == "__main__":
    DataPipeline().kickoff()

On dual-branch merging

The example listens to only sample_and_advise to keep the code simple; in production, merge both branches using the official or_(sample_and_advise, advise_directly) as the listen target — or_ means "trigger when any source completes," with a corresponding and_ (trigger only when all sources complete) — both exported from crewai.flow.flow.

24.5 Running Results and Extension Directions

mkdir -p data && python -c "
import pandas as pd, numpy as np
df = pd.DataFrame({'revenue': np.random.gamma(5, 200, 800),
                   'visits': np.random.poisson(30, 800)})
df.to_csv('data/sales.csv', index=False)"
python -m flow   # or crewai run flow

Typical log order confirms the orchestration logic: [load] 800 rows x 2 columns → takes the small branch → cleaning advice written to state as structured object → analyst alternates between the two stat tools → report produced. Directions for further evolution:

• Checkpoint persistence: add @persist to the Flow or Chapter 22's checkpoint mechanism; long analysis interruptions can resume:

from crewai.flow.persistence import persist


class DurablePipeline(DataPipeline):

    @persist  # After each method executes, automatically snapshot state to disk;
              # on re-run, restores to the latest snapshot and continues
    @listen(analyze)
    def build_report(self):
        report = make_report_crew().kickoff(inputs={
            "cleaning": self.state.cleaning.model_dump_json(indent=2),
            "stats": self.state.stats_summary,
        })
        self.state.final_report = report.raw

• More branch dimensions: in route_by_size, also branch on missing_ratio > 40% into a "heavy-cleaning" chain;
• API-ize: the Flow also applies to Chapter 22's kickoff + polling pattern; FlowState is naturally serializable into a job table.

Chapter Summary

• Flow selection criterion: when the process includes deterministic code steps or branching routing, Flow + Crew combination beats pure Crew;
• Pydantic root state + output_pydantic ensures steps pass validated structured objects between them;
• pandas capability wrapped as BaseTool (args_schema validation, compact string return) — LLM decides "which stat to call, how to interpret";
• @start/@router/@listen trio completes event-driven orchestration; decisions like size thresholds stay in Python code;
• Dual-branch merging uses or_/and_ to control trigger semantics.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. In this project, why is "deciding which branch based on row count" placed in ordinary Python code inside @router, rather than letting the Agent decide?

2. After setting output_pydantic=CleaningAdvice on a Task, where is the cleaning advice stored in the kickoff result?

3. When using pandas describe results as a tool return value, why convert to a compact string and limit row count?

4. After two mutually exclusive branches (large/small) both need to execute statistical analysis, what is the correct listen pattern?

🛠️ Hands-On Exercises

1. Add detection to probe_csv: find column names with missing rate exceeding 60%, store them in DatasetInfo.sparse_columns, and explicitly remind the cleaning advisor to focus on these columns in the task description.
2. Add a new OutlierTool: based on IQR (interquartile range) rules, return per-column outlier row numbers and count upper-bound summaries, so the analysis crew can cite outlier evidence in its report.
3. Add a third exit "dirty" to route_by_size (returned when missing_ratio > 0.4), write a "heavy-cleaning crew" branch for it, and merge all three branches into analyze using or_.

Final capstone: put Crew memory and scheduled execution to work, building a competitor monitoring intelligence system — Chapter 25: Capstone 3.