第 25 章 · 实战一：智能数据分析助手
本章目标：综合运用工具、Agent、Workflow 与 Team，从零构建一个"上传销售 CSV → 自动探查 → 多轮分析 → 生成图表 → 输出 Markdown 报告"的智能数据分析助手，把前 24 章的能力串成一条完整生产线。
25.1 需求与架构设计
业务方不想写 SQL，也不想学 pandas——他们希望用一句话拿到结论和图。我们把这条链路拆成四个角色：
用户问题："上季度哪个区域销售额下滑最严重？"
│
▼
┌─────────────────────────────────────────────┐
│ 报告 Workflow │
│ │
│ Step1 数据探查(函数) ──▶ Step2 深度分析(Agent) │
│ │ │ │
│ describe/分组聚合 调 query_csv │
│ 结果注入上下文 多轮查询+画图 │
│ │ │
│ Step3 报告撰写(Agent) │
└─────────────────────────────────────────────┘
│
▼
Markdown 分析报告 + charts/*.png分工原则与第 16 章一致：确定性的活给函数执行器（零 token），需要推理的活才交给 Agent。数据探查是固定动作（describe + 维度枚举），没必要烧模型；而"哪里的解读值得写进报告"必须由模型判断。
先准备一份示例数据 sales.csv：
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安装依赖：
pip install agno pandas matplotlib python-dotenv25.2 数据探查工具：query_csv
第一个自定义工具负责所有确定性数据访问。返回 dict 结构化结果而非拼接文本——第 8 章讲过，模型从结构化字段中提取信息比解析大段文字更可靠：
# data_tools.py —— 数据探查与图表生成两个工具
import os
import pandas as pd
from agno.tools import tool
CSV_PATH = "sales.csv"
@tool(name="query_csv", description="对销售 CSV 执行探查或分组聚合查询")
def query_csv(operation: str, group_by: str = "", metric: str = "") -> dict:
"""operation 支持 describe / group_sum / top_n；group_by 为分组列名"""
df = pd.read_csv(CSV_PATH)
if operation == "describe":
# 整表概览：列类型、缺失值、数值列统计
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":
# 分组聚合：如按 region 汇总 amount
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":
# 排名前 N 的明细行
top = df.nlargest(3, metric).to_dict(orient="records")
return {"top_by": metric, "rows": top}
return {"error": f"未知操作: {operation}"}三个要点：
@tool(name=..., description=...)显式命名并描述工具——这份描述会进入模型的函数定义，写得越具体，模型选对操作的命中率越高；- 参数签名即 JSON Schema 来源：
operation: str必填，group_by/metric带默认值可选； - 异常输入返回
{"error": ...}而不是抛异常，让模型有机会自行修正参数重试。
25.3 图表生成工具：chart
matplotlib 默认在无显示环境（服务器/Docker）下会报错，必须先切到 Agg 后端。生成的文件路径返回给模型，报告阶段直接引用：
# data_tools.py 续 —— 图表工具
import matplotlib
matplotlib.use("Agg") # 无显示环境必设，放在 pyplot 导入之前
import matplotlib.pyplot as plt
os.makedirs("charts", exist_ok=True)
@tool(name="chart", description="按分组绘制柱状图并保存 PNG，返回文件路径")
def chart(group_by: str, metric: str, title: str = "") -> dict:
"""读取 CSV 后按 group_by 列汇总 metric 列并出图"""
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) # 去掉多余边框更清爽
plt.tight_layout()
path = f"charts/{group_by}_{metric}.png"
fig.savefig(path, dpi=150)
plt.close(fig) # 及时释放，长会话不泄漏内存
return {"saved_to": path, "data": values.to_dict()}为什么返回路径而不是图片内容
模型无法"看"本地 PNG，但报告里需要引用它。返回 saved_to 路径让模型在 Markdown 里写出  即可；若要真正视觉理解图片，需走多模态输入（第 12 章），那是另一个场景。
25.4 分析 Agent 组装
现在把两个工具装到一个分析 Agent 上。指令里写明工作纪律：先探查再深挖、每个结论必须有数字支撑、关键对比配图：
# analyst.py —— 分析 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=[
"你是资深数据分析师。",
"回答任何问题前，先用 query_csv(operation='describe') 了解数据结构。",
"做趋势或对比判断时，先用 group_sum 取数，再决定是否需要 chart 出图佐证。",
"每个结论都必须附带来自工具的具体数字，禁止凭空推断。",
"最终回复用中文 Markdown 组织：结论先行，证据在后。",
],
markdown=True,
)
if __name__ == "__main__":
analyst.print_response(
"分析这份销售数据的区域表现，指出最值得关注的信号，并给出一张支撑图",
markdown=True,
)运行观察事件流（第 4 章的技巧）能看到典型行为序列：describe → 两三次 group_sum 对比 → 一次 chart → 组织语言作答。多轮工具调用完全由模型自主决策，这正是 Agno 工具循环的价值：你只定义能力边界，执行路径交给模型。
# observe_events.py —— 用事件流观察 Agent 的工具调用序列
from agno.agent import RunEvent
stream = analyst.run(
"对比各区域销售额，指出最强与最弱的市场", stream=True, stream_events=True
)
for chunk in stream:
if chunk.event == RunEvent.tool_call_started:
# 打印模型自主选择的工具与参数，验证它是否按指令先探查后深挖
print(f"[调用] {chunk.tool.tool_name}: {chunk.tool.tool_args}")25.5 报告 Workflow：三步流水线
单独问答够用了，但业务方要的是固定格式的周报。用 Workflow 把流程固化——Step1 函数探查（免费）、Step2 Agent 深度分析、Step3 Agent 报告撰写：
# report_pipeline.py —— 三步报告流水线
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 函数执行器：确定性探查，零 token 成本"""
profile = query_csv(operation="describe")
return StepOutput(content=f"数据概况（自动生成）：\n{profile}")
deep_analyst = Agent(
name="DeepAnalyst",
model=model,
tools=[query_csv, chart], # 深度分析仍可继续取数、补图
instructions=[
"基于上游提供的数据概况，围绕用户关注点完成深度分析。",
"至少进行两次分组对比，并为最重要的发现生成一张柱状图。",
"输出：核心发现（带数字）+ 图表路径列表。",
],
)
writer = Agent(
name="Writer",
model=model,
instructions=[
"把上游的分析结论改写成面向管理层的 Markdown 周报。",
"固定结构：# 本周销售分析 / ## 核心结论（≤3条）/ ## 详细解读 / ## 建议。",
"正文中以 Markdown 图片语法引用上游给出的图表路径。",
"不要出现'根据数据显示'之类的套话，直接陈述事实。",
],
)
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(
"重点关注各区域的环比变化与潜在风险", markdown=True
)注意 probe 的签名约定：接收 StepInput、返回 StepOutput，通过 previous_step_content 在步骤间传递数据（第 20 章的模式）。整条流水线只有 Step2 会消耗 token，且它的起点已经是结构化的数据画像——比裸问模型省得多。
25.6 可选增强：Team 双角色打磨报告
如果报告质量要求高（对外发布级别），可以把 Step2~Step3 换成一个 Team：分析师产出初稿，审稿人逐条核对数字后定稿。Team 的路由模式让审稿意见自动回传（详见第 15 章）：
# team_report.py —— 分析师 + 审稿人协作（替换 workflow 的后两步）
from agno.team import Team
fact_checker = Agent(
name="FactChecker",
model=model,
tools=[query_csv],
instructions=["核对报告中每个数字，发现与 CSV 不符立即指出并给出正确值。"],
)
report_team = Team(
name="report-team",
model=model,
members=[deep_analyst, fact_checker, writer],
instructions=[
"协作产出周报：Analyst 出分析稿，FactChecker 核对全部数字，Writer 定稿。",
"存在未通过的核对意见时不得进入终稿。",
],
)
report_team.print_response(
"基于数据概况：……（probe 步骤输出）\n请产出本周销售周报",
markdown=True,
)实战取舍建议：内部周报用 25.5 的线性 Workflow（快、便宜、稳定）；对外发布再上 Team（慢一点，但数字经过核验）。不要默认全上 Team——多一个成员就多几轮交互 token。
25.7 本章小结
- 综合实战的架构原则：确定性步骤用函数执行器，推理步骤用 Agent，格式固化用 Workflow；
- 工具返回结构化 dict 并显式声明 name/description，能显著提升模型调用准确率；
- matplotlib 在服务端必须切
Agg后端；图表返回文件路径供报告引用； - 线性 Workflow 与 Team 的取舍：成本与质量的天平，按受众选择而非无脑堆成员；
- 这条"探查 → 分析 → 撰写"流水线可直接迁移到你自己的业务表上——换掉 CSV_PATH 和提示词即可。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 为什么 Step1「数据探查」用函数执行器而不是 Agent？
2. matplotlib 在无显示的服务器环境中绘图，代码必须先做什么？
3. 关于 query_csv 工具的设计，下列做法最合理的是？
4. 内部周报场景下，相比把 Writer/FactChecker 都放进 Team 协作，选择三步线性 Workflow 的主要理由是？
🛠️ 动手实践
- 给
query_csv增加一个filter操作：支持按列值过滤后再聚合（如只看华东区），并让分析 Agent 实际用到它。 - 把 25.5 流水线的 Step3 输出落盘为
report.md文件，并用 Python 定时任务（schedule库）模拟每周一自动生成周报。 - 参考 25.6 把
fact_checker加入现有 Workflow 作为第四个 Step，对比加入前后报告中的数字错误率，写下你的取舍结论。
把这套"探查 → 分析 → 撰写"骨架换成你的业务数据试试。下一章我们把多路检索交给 Team 协作——第 26 章：多源研究助手 Team。