Chapter 26 · Capstone 1: Intelligent Data Analysis Assistant

Learning objectives: Synthesize tools, Agents, Workflows, and Teams from scratch to build an "upload sales CSV → auto-probe → multi-turn analysis → generate charts → produce Markdown report" intelligent data analysis assistant, tying together the capabilities from all 24 preceding chapters into a single production line.

26.1 Requirements and Architecture Design

The business side doesn't want to write SQL or learn pandas — they want answers and charts in a single sentence. We break this chain into four roles:

User question: "Which region had the steepest sales drop last quarter?"
    │
    ▼
┌─────────────────────────────────────────────┐
│  Report Workflow                             │
│                                              │
│  Step1 Data probe (function) ──▶ Step2 Deep analysis (Agent) │
│         │                        │           │
│   describe / group aggregation   call query_csv  │
│   result injected into context   multi-turn query + charting   │
│                                  │           │
│                    Step3 Report writing (Agent)     │
└─────────────────────────────────────────────┘
    │
    ▼
Markdown analysis report + charts/*.png

The division-of-labor principle matches Chapter 21: deterministic work goes to function executors (zero tokens); only reasoning work goes to Agents. Data probing is a fixed action (describe + dimension enumeration) — no need to burn model tokens; but "which insights are worth writing into the report" must be judged by the model.

Prepare a sample dataset sales.csv first:

month,region,product,amount,quantity
2025-07,华东,A产品,120500,310
2025-07,华南,B产品,88000,220
2025-08,华东,B产品,96000,240
2025-08,西南,A产品,45000,150
2025-09,华东,A产品,132000,330
2025-09,华南,C产品,41000,95
2025-09,西南,B产品,39000,105

Install dependencies:

pip install agno pandas matplotlib python-dotenv

26.2 Data Probing Tool: query_csv

The first custom tool handles all deterministic data access. Return a structured dict, not concatenated text — as covered in Chapter 8, the model extracts information from structured fields far more reliably than parsing long prose:

# data_tools.py —— Two tools: data probe and chart generation
import os
import pandas as pd
from agno.tools import tool

CSV_PATH = "sales.csv"

@tool(name="query_csv", description="Run probe or group-aggregation queries on the sales CSV")
def query_csv(operation: str, group_by: str = "", metric: str = "") -> dict:
    """operation supports describe / group_sum / top_n; group_by is the column name to group by"""
    df = pd.read_csv(CSV_PATH)

    if operation == "describe":
        # Whole-table overview: column types, missing values, numeric summaries
        return {
            "shape": list(df.shape),
            "columns": list(df.columns),
            "dtypes": {c: str(t) for c, t in df.dtypes.items()},
            "missing": int(df.isna().sum().sum()),
            "numeric_summary": df.describe().to_dict(),
        }

    if operation == "group_sum":
        # Group aggregation: e.g., sum amount by region
        grouped = df.groupby(group_by)[metric].sum().sort_values(ascending=False)
        return {"group_by": group_by, "metric": metric,
                "values": grouped.to_dict()}

    if operation == "top_n":
        # Top-N detail rows
        top = df.nlargest(3, metric).to_dict(orient="records")
        return {"top_by": metric, "rows": top}

    return {"error": f"Unknown operation: {operation}"}

Three design notes:

• @tool(name=..., description=...) explicitly names and describes the tool — this description enters the model's function definition; the more specific it is, the higher the hit rate when the model picks the right operation;
• The parameter signature is the JSON Schema source: operation: str is required; group_by / metric have defaults and are optional;
• Invalid input returns {"error": ...} rather than raising — letting the model self-correct and retry.

26.3 Chart Generation Tool: chart

matplotlib defaults to an interactive backend that errors in headless environments (servers / Docker). You must switch to the Agg backend first. The generated file path is returned to the model so the report stage can reference it directly:

# data_tools.py continued —— Chart tool
import matplotlib
matplotlib.use("Agg")          # Must set before importing pyplot in headless envs
import matplotlib.pyplot as plt

os.makedirs("charts", exist_ok=True)

@tool(name="chart", description="Draw a bar chart grouped by column and save as PNG; returns file path")
def chart(group_by: str, metric: str, title: str = "") -> dict:
    """Read CSV, aggregate metric by group_by column, and produce a chart"""
    df = pd.read_csv(CSV_PATH)
    values = df.groupby(group_by)[metric].sum()

    fig, ax = plt.subplots(figsize=(7, 4))
    ax.bar(values.index.astype(str), values.values, color="#0969da")
    ax.set_title(title or f"{metric} by {group_by}")
    ax.spines[["top", "right"]].set_visible(False)   # Cleaner look without extra borders
    plt.tight_layout()

    path = f"charts/{group_by}_{metric}.png"
    fig.savefig(path, dpi=150)
    plt.close(fig)                                   # Release promptly; avoids memory leaks in long sessions
    return {"saved_to": path, "data": values.to_dict()}

Why return the path instead of the image content

The model can't "see" local PNG files, but the report needs to reference them. Returning the saved_to path lets the model write ![...](charts/region_amount.png) in Markdown; true visual understanding requires multimodal input (Chapter 13), which is a separate scenario.

26.4 Assembling the Analysis Agent

Now equip the analysis Agent with both tools. Instructions spell out working discipline: probe first, then dig deeper; every conclusion must be backed by numbers; key comparisons get charts:

# analyst.py —— Analysis Agent
import os
from dotenv import load_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from data_tools import query_csv, chart

load_dotenv()
model = OpenAIChat(
    id="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
)

analyst = Agent(
    name="Analyst",
    model=model,
    tools=[query_csv, chart],
    instructions=[
        "You are a senior data analyst.",
        "Before answering any question, call query_csv(operation='describe') to understand the data structure first.",
        "When making trend or comparison judgments, fetch numbers with group_sum first, then decide whether a chart is needed.",
        "Every conclusion must be backed by specific numbers from the tool; do not speculate.",
        "Organize the final reply in Chinese Markdown: conclusions first, evidence after.",
    ],
    markdown=True,
)

if __name__ == "__main__":
    analyst.print_response(
        "Analyze regional performance in this sales data, point out the most noteworthy signal, and produce a supporting chart",
        markdown=True,
    )

Observe the event stream (the trick from Chapter 4) and you'll see the typical sequence: describe → two or three group_sum comparisons → one chart → organized prose. Multi-turn tool calls are entirely model-driven — this is exactly the value of Agno's tool loop: you define the capability boundary, and the execution path is left to the model.

# observe_events.py —— Watch the Agent's tool-call sequence via the event stream
from agno.agent import RunEvent

stream = analyst.run(
    "Compare sales across regions; identify the strongest and weakest markets", stream=True, stream_events=True
)
for chunk in stream:
    if chunk.event == RunEvent.tool_call_started:
        # Print the tool and args the model chose autonomously; verify it probes first then digs deeper
        print(f"[call] {chunk.tool.tool_name}: {chunk.tool.tool_args}")

26.5 Report Workflow: A Three-Step Pipeline

Single Q&A is enough for ad-hoc questions, but the business side wants a fixed-format weekly report. Use a Workflow to hardcode the flow — Step1 function probe (free), Step2 Agent deep analysis, Step3 Agent report writing:

# report_pipeline.py —— Three-step report pipeline
from agno.agent import Agent
from agno.workflow import Step, StepInput, StepOutput, Workflow
from data_tools import query_csv
from analyst import analyst, model


def probe(step_input: StepInput) -> StepOutput:
    """Step1 function executor: deterministic probe at zero token cost"""
    profile = query_csv(operation="describe")
    return StepOutput(content=f"Data profile (auto-generated):\n{profile}")


deep_analyst = Agent(
    name="DeepAnalyst",
    model=model,
    tools=[query_csv, chart],          # Deep analysis can still fetch and chart
    instructions=[
        "Based on the data profile from upstream, complete an in-depth analysis around the user's focus.",
        "Perform at least two group comparisons, and produce a bar chart for the most important finding.",
        "Output: key findings (with numbers) + a list of chart paths.",
    ],
)


writer = Agent(
    name="Writer",
    model=model,
    instructions=[
        "Rewrite the upstream analysis into a management-oriented Markdown weekly report.",
        "Fixed structure: # Weekly Sales Analysis / ## Key Conclusions (≤3) / ## Detailed Interpretation / ## Recommendations.",
        "Reference chart paths from upstream using Markdown image syntax in the body.",
        "Avoid boilerplate phrases like 'according to the data'; state facts directly.",
    ],
)


workflow = Workflow(
    name="sales-weekly-report",
    steps=[
        Step(name="probe", executor=probe),
        Step(name="analyze", executor=deep_analyst),
        Step(name="write", executor=writer),
    ],
)

if __name__ == "__main__":
    workflow.print_response(
        "Focus on quarter-over-quarter regional changes and potential risks", markdown=True
    )

Note the probe signature convention: accepts StepInput, returns StepOutput, passing data between steps via previous_step_content (the pattern from Chapter 11). Only Step2 consumes tokens along the entire pipeline, and its starting point is already a structured data profile — far cheaper than asking the model blind.

26.6 Optional Enhancement: Team for Polished Reports

If report quality must be publication-grade, replace Step2–Step3 with a Team: the analyst produces a draft, the fact-checker verifies every number line-by-line, then the writer finalizes. The Team's router mode loops feedback automatically (see Chapter 9):

# team_report.py —— Analyst + Fact-checker collaboration (replaces last two workflow steps)
from agno.team import Team

fact_checker = Agent(
    name="FactChecker",
    model=model,
    tools=[query_csv],
    instructions=["Verify every number in the report; flag any mismatch with the CSV and provide the correct value immediately."],
)

report_team = Team(
    name="report-team",
    model=model,
    members=[deep_analyst, fact_checker, writer],
    instructions=[
        "Collaborate on the weekly report: Analyst produces the draft, FactChecker checks all numbers, Writer finalizes.",
        "Do not enter the final draft while any verification意见 is outstanding.",
    ],
)

report_team.print_response(
    "Based on the data profile: … (output from the probe step)\nPlease produce this week's sales weekly report",
    markdown=True,
)

Practical guidance: internal weekly reports use the linear Workflow from 26.5 (fast, cheap, stable); external publications go with the Team (slightly slower, but numbers are verified). Don't default to Team — each additional member adds interaction rounds and token cost.

26.7 Chapter Summary

• Architecture principle for capstone projects: deterministic steps → function executor; reasoning steps → Agent; fixed formatting → Workflow;
• Tools returning structured dicts with explicit name/description significantly boost the model's call accuracy;
• matplotlib on servers must switch to the Agg backend; charts return file paths for report referencing;
• Trade-off between linear Workflow and Team: balance cost vs. quality based on audience, not by blindly stacking members;
• This "probe → analyze → write" skeleton is directly portable to your own business tables — just swap CSV_PATH and the prompts.

🧪 随堂测验

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

1. Why does Step1 "data probing" use a function executor rather than an Agent?

2. What must you do first in matplotlib code when drawing in a headless server environment?

3. Regarding query_csv tool design, which approach is most reasonable?

4. For an internal weekly report scenario, compared to putting both Writer and FactChecker in a Team, the main reason to choose a three-step linear Workflow is?

🛠️ Hands-On Exercises

1. Add a filter operation to query_csv: support filtering by column values before aggregation (e.g., view only the 华东 region), and have the analysis Agent actually use it.
2. Make Step3's output of the 26.5 pipeline write to report.md; simulate weekly auto-generation with a Python scheduled task (schedule library).
3. Following 26.6, add fact_checker to the existing Workflow as a fourth Step; compare the number-error rate in reports before and after, and write down your trade-off conclusion.

Try swapping this "probe → analyze → write" skeleton with your own business data. Next chapter we hand multi-source retrieval to Team collaboration — Chapter 27: Multi-Source Research Assistant Team.