已开源本文涉及的代码到ai-demo，欢迎star~

一、引言

在日常的数据分析工作中，我们经常会遇到这样的场景：

“这里有一个 CSV 文件，帮我分析一下数据趋势，生成可视化图表，再写一份分析报告。”

传统做法需要手动编写 Pandas 脚本 → Matplotlib 绘图 → 整理 Markdown 文档，整个过程繁琐且重复。

但如果把这条指令直接交给 AI Agent 呢？—— 这就是本文要探索的内容。

我们借助 DeepAgents（一个基于 LangGraph 构建的增强型 ReAct Agent 框架），只需 十几行核心代码，就能让 AI 自动完成：读取 CSV → 数据分析 → 绘制图表 → 输出分析报告 的全流程。

先看最终效果：

用户输入：分析 sales.csv，生成图表图片，写出 Markdown 报告
    ↓
Agent 自主完成：
  1. 读取 CSV 文件
  2. 用 Python 进行数据统计
  3. 调用 matplotlib 绘制柱状图/折线图
  4. 将分析结果写入 sales_analysis.md

---

二、DeepAgents 是什么？

DeepAgents 是一个基于 LangGraph 构建的增强型 ReAct Agent 框架。相比传统的 LangChain Agent，它有几个突出的特点：

特性	说明
🧠 增强型 ReAct	内置了更强大的工具调用与推理循环，支持复杂多步任务
🔧 零配置工具	无需额外配置 Skill、Tool、Middleware，Agent 自主发现和使用能力
🖥️ 本地 Shell 后端	可在本地执行 Python 脚本、Shell 命令，天然具备代码生成与执行能力
📎 文件上传	支持在对话前上传文件（CSV、图片等），Agent 可直接读取操作
🔁 流式输出	基于 astream_events 支持实时流式响应

简单来说：你给它一个任务，它自己想办法完成，过程中可以写代码、读文件、调 API、生成产物。

---

三、实战代码逐行解析

下面是我们本次演示的核心代码，我将 逐段拆解 每一部分的作用。

3.1 环境准备与依赖引入

from pathlib import Path
from deepagents.backends import LocalShellBackend
from dotenv import load_dotenv

_root = Path(__file__).resolve().parent
load_dotenv(_root / ".env")

from common import get_deepseek_model, get_minimax_model, print_messages, run_and_print_stream

说明：

• LocalShellBackend：DeepAgents 的核心后端，提供一个本地 Shell 环境，Agent 可以在此执行 Python 脚本、Shell 命令。
• load_dotenv：加载 .env 配置文件（通常存放 API Key 等敏感信息），这是 deepagents 的推荐实践。
• common 模块：项目中封装的辅助函数，用于获取 LLM 模型实例、打印消息、运行流式输出。
• get_deepseek_model()：返回 DeepSeek 大模型实例，作为 Agent 的推理引擎。

为什么要用 LocalShellBackend？—— 因为它让 Agent 具备了在真实环境中执行代码的能力，这是 Agent 完成数据分析任务的基石。

3.2 初始化后端

backend = LocalShellBackend(
    root_dir=_root,
    inherit_env=True,
    env=None,
    virtual_mode=False
)

参数含义：

参数	值	作用
root_dir	脚本所在目录	Agent 的工作目录，所有文件操作基于此路径
inherit_env	True	继承当前进程的环境变量（如 PATH、Python 路径）
virtual_mode	False	关闭虚拟模式，Agent 直接在真实 Shell 中执行命令

3.3 准备样本数据

import csv
import io

data = [
    ["date", "product", "sold", "revenue"],
    ["2026-01-01", "A", 10, 100],
    ["2026-01-02", "B", 5, 50],
    ["2026-01-03", "C", 20, 200],
    ["2026-01-04", "A", 15, 150],
    ["2026-01-05", "B", 10, 100],
]

text_buf = io.StringIO()
writer = csv.writer(text_buf)
writer.writerows(data)
csv_bytes = text_buf.getvalue().encode("utf-8")
text_buf.close()

backend.upload_files([("data/sales.csv", csv_bytes)])

这段代码做了两件事：

1. 构建 CSV 数据：使用 Python 内置的 csv 模块在内存中生成一份销售数据，包含日期、产品、销量、收入四个字段，共 5 行记录。
2. 上传到后端：backend.upload_files() 将 CSV 文件上传到 Agent 工作目录的 data/sales.csv 路径下。

💡 upload_files 是 LocalShellBackend 提供的核心方法，它让 Agent 在对话开始前就拥有数据文件，而无需用户手动放置。

样本数据预览：

date	product	sold	revenue
2026-01-01	A	10	100
2026-01-02	B	5	50
2026-01-03	C	20	200
2026-01-04	A	15	150
2026-01-05	B	10	100

3.4 创建 Agent

import uuid
from langgraph.checkpoint.memory import InMemorySaver
from deepagents import create_deep_agent

checkpointer = InMemorySaver()

agent = create_deep_agent(
    model=get_deepseek_model(),
    backend=backend,
    checkpointer=checkpointer
)

thread_id = str(uuid.uuid4())
config = {
    "configurable": {"thread_id": thread_id},
    "recursion_limit": 999
}

关键点解读：

• InMemorySaver：LangGraph 的检查点存储器，用于保存对话的中间状态。这使得 Agent 可以在多轮交互中保持上下文，也支持 人机交互（Human-in-the-Loop）—— 即 Agent 在执行关键操作前可以请求用户确认。
• create_deep_agent()：DeepAgents 的核心工厂函数，传入模型和后端，返回一个完整的 Agent 实例。注意这里不需要配置任何 Skill 或 Tool，Agent 会自主发现可用能力。
• recursion_limit: 999：Agent 任务较复杂时（如数据分析涉及多步推理、代码执行），默认的 25 步递归限制容易触发。推荐设置为较大的值，给 Agent 足够的推理空间。

3.5 定义任务指令

input_messages = [
    {
        "role": "system",
        "content": "You are a helpful assistant, please response in Chinese."
    },
    {
        "role": "user",
        "content": (
            "分析当前目录下的data子目录的sales.csv，"
            "然后生成一个友好的plot的png图片"
            "（图片里的文字用Heiti TC字体），"
            "完成后，把分析结果写到./data/sales_analysis.md"
        )
    },
]

这条 prompt 包含了三个子任务：

1. 读取并分析数据 → data/sales.csv
2. 生成可视化图片 → 指定 Heiti TC 字体（macOS 上的中文字体），避免中文乱码
3. 输出分析报告 → 写入 ./data/sales_analysis.md

Agent 需要自主规划执行顺序：先读文件 → 数据分析 → 绘图 → 写报告，中间可能涉及多次代码生成与调试。

3.6 启动运行

if __name__ == "__main__":
    import asyncio
    asyncio.run(run_and_print_stream(agent, input_messages, config))

run_and_print_stream 是一个异步辅助函数，它调用 agent.astream_events() 获取 Agent 的实时事件流，并将中间步骤（思考过程、工具调用、执行结果）打印到控制台。这是 DeepAgents 流式输出 的核心体验。

---

四、完整可运行代码

将以上所有片段整合，得到完整的 analyze_csv.py：

from pathlib import Path
from deepagents.backends import LocalShellBackend
from dotenv import load_dotenv
import csv, io, uuid
from langgraph.checkpoint.memory import InMemorySaver
from deepagents import create_deep_agent

_root = Path(__file__).resolve().parent
load_dotenv(_root / ".env")
from common import get_deepseek_model, print_messages, run_and_print_stream

# 初始化后端
backend = LocalShellBackend(
    root_dir=_root,
    inherit_env=True,
    virtual_mode=False
)

# 准备样本 CSV 数据
data = [
    ["date", "product", "sold", "revenue"],
    ["2026-01-01", "A", 10, 100],
    ["2026-01-02", "B", 5, 50],
    ["2026-01-03", "C", 20, 200],
    ["2026-01-04", "A", 15, 150],
    ["2026-01-05", "B", 10, 100],
]
text_buf = io.StringIO()
csv.writer(text_buf).writerows(data)
backend.upload_files([("data/sales.csv", text_buf.getvalue().encode("utf-8"))])
text_buf.close()

# 创建 Agent
checkpointer = InMemorySaver()
agent = create_deep_agent(
    model=get_deepseek_model(),
    backend=backend,
    checkpointer=checkpointer
)

config = {
    "configurable": {"thread_id": str(uuid.uuid4())},
    "recursion_limit": 999
}

input_messages = [
    {"role": "system", "content": "You are a helpful assistant, please response in Chinese."},
    {"role": "user", "content": "分析当前目录下的data子目录的sales.csv，然后生成一个友好的plot的png图片（图片里的文字用Heiti TC字体），完成后，把分析结果写到./data/sales_analysis.md"},
]

if __name__ == "__main__":
    import asyncio
    asyncio.run(run_and_print_stream(agent, input_messages, config))

---

五、运行效果与流程拆解

当执行上述代码后，Agent 会经历以下步骤（通过流式输出可以实时看到）：

Step 1: 思考 → "我需要先读取 sales.csv 了解数据结构和内容"
Step 2: 执行 → cat data/sales.csv
Step 3: 思考 → "数据包含 date/product/sold/revenue 四列，共5行"
Step 4: 执行 → 编写 Python 脚本进行统计分析（总销量、总收入、按产品汇总等）
Step 5: 执行 → 调用 matplotlib 绘制图表，设置 Heiti TC 字体
Step 6: 执行 → 编写 Markdown 报告，写入 data/sales_analysis.md
Step 7: 完成 → 返回执行结果

最终产出的文件结构：

project_root/
├── data/
│   ├── sales.csv              # 原始数据（Agent 输入）
│   ├── sales_analysis.md      # 分析报告（Agent 输出）
│   └── *.png                  # 可视化图表（Agent 输出）
├── analyze_csv.py             # 本示例代码
└── .env                       # 环境变量（API Key 等）

---

六、DeepAgents 核心机制解读

为什么 DeepAgents 能做到"零配置"？

6.1 增强型 ReAct 循环

与普通 ReAct Agent 不同，DeepAgents 的推理循环做了大量增强：

用户输入
    ↓
Agent 思考（Thought）→ 我需要做什么
    ↓
Agent 行动（Action）→ 调用工具（执行代码、读文件等）
    ↓
观察结果（Observation）→ 代码输出/报错信息
    ↓
Agent 再次思考 → 结果符合预期吗？下一步做什么？
    ↓
... 循环直到任务完成

这个循环是 自适应 的 —— 如果代码报错，Agent 会分析错误信息并修正代码重新执行，无需人工干预。

6.2 LocalShellBackend 的秘密

LocalShellBackend 是整个能力的基石，它提供了：

能力	说明
exec_command	执行任意 Shell 命令或 Python 脚本
read_file / write_file	读写工作目录下的文件
upload_files	在对话前上传文件
list_directory	浏览目录结构

Agent 通过 backend 间接拥有了操作本地文件系统的能力，这是它能自主完成数据分析任务的根本原因。

6.3 与 LangGraph Checkpointer 的配合

InMemorySaver 作为检查点存储器，实现了：

• 状态持久化：Agent 的每一步推理、每个工具调用结果都被保存
• 断点恢复：支持在任意步骤中断并手动干预（Human-in-the-Loop）
• 调试回放：可以回放 Agent 的执行过程，便于排查问题

---

七、扩展应用场景

基于同样的模式，你可以快速实现更多自动化场景：

场景	修改点
📊 金融数据报表	替换为股票/基金 CSV，prompt 增加技术指标计算要求
🏭 生产质量分析	上传质检 CSV，要求生成控制图和异常检测报告
📈 竞品监控周报	配置定时任务，自动拉取数据 → 分析 → 生成周报
🛒 电商销售分析	分析订单 CSV，生成用户画像、商品关联分析等
📝 日志异常检测	上传服务器日志，自动识别异常模式并输出诊断报告

只需更换 CSV 文件和调整 prompt 描述，Agent 就能自动适配新的分析需求，无需修改任何代码逻辑。

---

八、结语

本文通过一个真实可运行的示例，展示了 DeepAgents + DeepSeek + LocalShellBackend 组合的强大之处：

只需一个自然语言指令 + 一个 CSV文件，AI Agent 就能自主完成从数据读取到报告输出的完整数据分析流程。

这种方式极大地降低了数据分析的门槛——你不需要写 Pandas、不需要调 Matplotlib、不需要组织报告格式，甚至不用手写skills，只需关注业务问题本身。

当然，DeepAgents 的能力远不止于此。它还可以接入 Web 搜索、API 调用、数据库查询等更多后端，构建更复杂的智能体系统。

---