构建数据分析 Agent

构建一个能够分析数据文件、生成可视化图表并分享结果的 Agent。

概述

本指南演示如何使用 Deep Agent（深度 Agent）构建一个数据分析 Agent。数据分析任务通常需要规划、代码执行以及与脚本、报告和图表等构建产物打交道——这些正是 Deep Agent 设计之初就具备的能力。

我们将构建的 Agent 能够：

1. 接受 CSV 文件进行分析
2. 执行探索性数据分析并生成可视化图表
3. 将结果分享到 Slack 频道

提示： Slack 集成是可选的。Agent 可以改为将构建产物保存在本地，或通过其他渠道分享结果。

关键概念

本教程涵盖：

• Backends（后端）——用于沙箱化代码执行
• 自定义 Tools（工具）——用于外部集成

---

环境准备

安装

安装核心依赖：

pip install deepagents

可选依赖

本教程使用以下可选组件：

• Slack Python SDK——用于分享结果（需要设置 Token）
• 用于代码执行的沙箱环境。详见可用提供商的安装说明

pip install slack-sdk

注意： 这些服务是可选的，但强烈建议在生产环境中使用沙箱环境。你也可以使用本地 Shell 后端（但需了解其安全注意事项），或直接从后端下载构建产物。

LangSmith

你用 LangChain 构建的应用通常包含多个步骤和多次 LLM 调用。随着应用日趋复杂，能够检查链或 Agent 内部究竟发生了什么变得至关重要。最佳方式是使用 LangSmith。

注册后，设置环境变量以开始记录追踪：

export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."

或者在 Python 中设置：

import getpass
import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()

---

设置后端

Deep Agent 使用 Backends（后端）在沙箱化环境中执行代码。

详见可用提供商的安装说明。

Daytona（推荐）

pip install langchain-daytona

from daytona import Daytona
from langchain_daytona import DaytonaSandbox

sandbox = Daytona().create()
backend = DaytonaSandbox(sandbox=sandbox)

验证沙箱是否就绪：

result = backend.execute("echo ready")
print(result)
# ExecuteResponse(output='ready', exit_code=0, ...)

Modal

import modal
from langchain_modal import ModalSandbox

app = modal.App.lookup("your-app")
modal_sandbox = modal.Sandbox.create(app=app)
backend = ModalSandbox(sandbox=modal_sandbox)

Runloop

pip install langchain-runloop

from runloop_api_client import RunloopSDK
from langchain_runloop import RunloopSandbox

api_key = "..."
client = RunloopSDK(bearer_token=api_key)
devbox = client.devbox.create()
backend = RunloopSandbox(devbox=devbox)

AgentCore（AWS Bedrock）

pip install langchain-agentcore-codeinterpreter

from bedrock_agentcore.tools.code_interpreter_client import CodeInterpreter
from langchain_agentcore_codeinterpreter import AgentCoreSandbox

interpreter = CodeInterpreter(region="us-west-2")
interpreter.start()
backend = AgentCoreSandbox(interpreter=interpreter)

本地 Shell

警告： 此后端提供不受限制的文件系统和 Shell 访问权限。仅在受控环境中用于开发和测试。详见安全注意事项。

from deepagents.backends import LocalShellBackend

backend = LocalShellBackend(root_dir=".", env={"PATH": "/usr/bin:/bin"})

---

上传样本数据

创建示例销售数据并上传到后端：

import csv
import io

# 创建示例销售数据
data = [
    ["Date", "Product", "Units Sold", "Revenue"],
    ["2025-08-01", "Widget A", 10, 250],
    ["2025-08-02", "Widget B", 5, 125],
    ["2025-08-03", "Widget A", 7, 175],
    ["2025-08-04", "Widget C", 3, 90],
    ["2025-08-05", "Widget B", 8, 200],
]

# 转换为 CSV 字节
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([("/home/daytona/data/sales_data.csv", csv_bytes)])

---

实现自定义工具

数据分析任务可能会产生报告或图表等构建产物。以下简单的工具通过 backend.download_files 下载这些文件，然后使用 Slack SDK 上传。我们也可以让 Agent 列出相关文件路径而非直接上传，以便相关人员根据需要单独获取。

from langchain.tools import tool
from slack_sdk import WebClient

slack_token = os.environ["SLACK_USER_TOKEN"]
slack_client = WebClient(token=slack_token)


@tool(parse_docstring=True)
def slack_send_message(text: str, file_path: str | None = None) -> str:
    """发送消息，可选包含附件（如图片）。

    Args:
        text: (str) 消息内容
        file_path: (str) 文件系统中附件的路径
    """
    if not file_path:
        slack_client.chat_postMessage(channel=channel, text=text)
    else:
        fp = backend.download_files([file_path])
        slack_client.files_upload_v2(
            channel="C0123456ABC",  # 替换为你自己的频道 ID
            content=fp[0].content,
            initial_comment=text,
        )

    return "Message sent."

注意： 通常好的做法是避免将凭据和其他秘密信息放入沙箱。此处我们通过工具在沙箱外部管理 Slack Token。

---

运行 Agent

实例化 Agent：

from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver
from deepagents import create_deep_agent

checkpointer = InMemorySaver()

agent = create_deep_agent(
    model="google_genai:gemini-3.5-flash",
    tools=[slack_send_message],
    backend=backend,
    checkpointer=checkpointer,
)

thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}

上述代码包含了：

• 选择的模型
• 自定义工具
• 后端
• 用于支持多轮对话的检查点

现在调用 Agent：

input_message = {
    "role": "user",
    "content": (
        "Analyze ./data/sales_data.csv in the current dir and generate a beautiful plot. "
        "When finished, send your analysis and the plot to Slack using the tool."
    ),
}

for step in agent.stream(
    {"messages": [input_message]},
    config,
    stream_mode="updates",
):
    for _, update in step.items():
        if update and (messages := update.get("messages")) and isinstance(messages, list):
            for message in messages:
                message.pretty_print()

对话过程输出示例

Agent 收到请求后，会逐步完成以下操作：

1. 发现数据文件——先尝试读取 ./data/sales_data.csv，发现路径不对，然后通过 pwd 和 ls 探索目录结构，最终在 /home/daytona/data/sales_data.csv 找到文件
2. 读取并理解数据——读取 CSV 文件内容，了解数据结构（日期、产品、销量、收入）
3. 编写分析脚本——自动生成一个完整的 Python 分析脚本（analyze_sales.py），使用 pandas、matplotlib 和 seaborn 进行数据处理和可视化
4. 生成可视化仪表板——执行脚本，生成包含 6 个子图的综合销售分析仪表板：
   • 每日收入柱状图
   • 每日销量柱状图
   • 各产品收入占比饼图
   • 各产品总收入水平柱状图
   • 各产品总销量水平柱状图
   • 各产品销售笔数分布饼图
5. 分享到 Slack——调用自定义的 slack_send_message 工具，将分析文本和生成的图表上传到指定的 Slack 频道

Agent 的自动探索和错误修正能力是其核心优势——当首次读取文件失败时，它会自主排查目录结构并纠正路径。

输出结果

================================== Ai Message ==================================
[{'text': "I'll help you analyze the sales data and create a beautiful plot, then send the results to Slack. Let me start by exploring the data.", ...}]

...（Agent 逐步发现文件、编写分析脚本、执行分析）...

================================== Ai Message ==================================
[{'text': 'The analysis has been completed. Let me share the results on Slack...'}, ...]

Tool Calls:
 slack_send_message (...)
 Args:
   text: "Sales Data Analysis Report\n..."
   file_path: /home/daytona/sales_analysis_plot.png

Agent 会将文本报告和可视化图表一起发送到 Slack 频道，团队成员可以直接在频道中查看分析结果。

---

下一步

• 了解如何自定义 Deep Agent——修改模型、工具集和行为
• 探索不同的沙箱提供商以适配你的基础设施
• 学习如何与 LangChain 生态系统中的其他工具集成
• 查看 Deep Agent 概念文档了解更多高级用法