本教程演示如何使用方舟 Managed Agents 完成一次端到端的数据分析：上传一份 CSV 文件，由 Agent 在云沙箱中完成数据读取、清洗、统计、图表生成，并最终输出一份自包含的 HTML 分析报告。
场景说明#
在企业的数据分析场景中，业务人员拿到 CSV 或 Excel 结构化数据后，往往只能看到原始记录，很难快速判断哪些信息值得关注。传统链路依赖分析师完成数据清洗、统计和可视化，响应周期长、重复性工作多。
这类场景适合交给 Managed Agents 处理：用户只需提供数据文件并描述分析目标，Agent 可以自动完成数据读取、清洗、代码执行、图表生成和结果总结，并输出一份结构清晰的分析报告。
| 环节 | 内容 |
|---|---|
| 输入 | 一个 CSV 文件，例如订单、销售、用户行为或运营数据 |
| 过程 | Agent 读取文件，使用 pandas 分析数据，使用 plotly 生成交互图表 |
| 输出 | 一个自包含的 report.html，包含摘要、关键指标、图表和行动建议 |
| 观测 | 通过 Session Events SSE 查看 Agent 消息、工具调用和状态变化 |
你将完成什么#
- 创建一个用于数据分析的 Environment。
- 创建一个可复用的数据分析 Agent。
- 上传 CSV 并挂载到 Session 沙箱。
- 发送分析任务并通过事件流观察执行过程。
- 查询 Session 产出的报告文件。
核心概念#
动手前先厘清方舟 Managed Agents 的四个基础对象。
| 概念 | 说明 |
|---|---|
| Agent | 一份可版本化、可复用的配置模板，封装模型、System Prompt 和工具集。创建后得到稳定的 agent_id，可被任意 Session 引用 |
| Environment | Agent 运行所在的云沙箱定义，决定网络策略与预装依赖，一次创建，多次复用 |
| Session | 一次具体的运行实例，把 Agent、Environment 和输入资源绑在一起。创建 Session 本身不会触发工作，需要主动下发消息 |
| Event | Session 内的追加式事件流。客户端用 user.message 下发任务，再通过 SSE 实时收到 Agent 的消息、工具调用与状态变化 |
工作流程#
前置条件#
- 已开通方舟 Managed Agents 和目标模型服务。
- 已创建 API Key，详情请参见 API Key 管理。
- 本地已安装
curl和jq。 - 准备一个 CSV 文件，例如
test_csv.csv。
将 API Key、Base URL 和常用配置配置为环境变量：
export ARK_API_KEY="你的方舟 API Key"
export ARK_BASE_URL="https://ark.cn-beijing.volces.com/api/v3"
export MODEL_ID="doubao-seed-2-1-pro-260628"
export TOOLSET_TYPE="agent_toolset_20260916"
export DATA_PATH="./test_csv.csv"
第 1 步：创建 Environment#
Environment 定义 Agent 的执行环境，例如网络访问权限、预装依赖和运行配置。本示例提前安装 pandas 和 plotly，Agent 执行任务时可以直接读取 CSV 文件、生成图表，无需运行时再安装依赖。Environment 创建完成后可以长期复用，后续多个 Session 都可以基于同一个 Environment 运行。
environment=$(curl -sS --fail-with-body "$ARK_BASE_URL/environments" \
-H "Authorization: Bearer $ARK_API_KEY" \
-H "Content-Type: application/json" \
-d @- <<'EOF'
{
"name": "data-analyst-env",
"description": "用于 CSV 数据分析报告生成的托管沙箱环境。",
"config": {
"type": "cloud",
"networking": {"type": "unrestricted"},
"packages": {
"pip": ["pandas", "plotly"]
}
}
}
EOF
)
ENVIRONMENT_ID=$(jq -er '.id' <<<"$environment")
echo "Environment: $ENVIRONMENT_ID"
第 2 步：定义分析 Agent#
Environment 创建完成后，创建 Agent。Agent 包含模型、System Prompt 和工具等配置，是 Session 执行任务时使用的核心资源。System Prompt 用于约束 Agent 的分析流程和输出格式，例如数据校验、统计分析、图表生成以及最终报告的组织方式。创建完成后，Agent 可以被多个 Session 复用。
先把 System Prompt 保存为变量：
ANALYST_SYSTEM_PROMPT='你是一名资深数据分析师，负责把 CSV 数据集转成可直接交付的 HTML 分析报告。
工作目标：
输出一份面向业务方可直接阅读的报告。报告应包含摘要、关键指标、图表、主要发现、数据质量说明和行动建议。
工作原则：
1. 先读取并理解原始 CSV 的结构，再开始分析。必须明确区分"原始输入字段"和"分析过程中派生出的字段"。
2. 所有关于数据规模、行列数、字段名、缺失值、重复值的描述，都必须基于原始 CSV，不得基于清洗后或派生后的 dataframe。
3. 大表先抽样查看，再按需全量聚合，避免无意义地打印整表。
4. 所有结论必须由具体数字支撑；不能只给判断，不给证据。
5. 使用 pandas 完成清洗与聚合，使用 plotly 生成交互式图表。
6. 生成至少 3 张图表，并在图表之间写简短说明，解释图表反映的业务洞察。
7. 为最终的报告生成一个 HTML 文件。
交付要求：
1. 生成可直接交付的 report.html，要求为单文件、自包含 HTML，可离线浏览，不依赖任何外部资源。
2. 报告内容至少包括：
- Executive Summary
- Key Metrics
- Data Quality Notes
- 3 个以上图表及对应洞察
- Action Recommendations
分析与表述要求：
1. 优先呈现最有行动价值的发现。
2. 建议必须区分"数据支持的结论"和"需要进一步验证的假设"。
3. 如果样本时间很短、缺少关键字段或无法支撑因果判断，必须明确写出限制。
4. 不得编造不存在的字段、维度、业务背景或结论。
执行要求：
1. 读取原始 CSV 后，先记录：
- 原始行数
- 原始列数
- 原始列名
- 缺失值情况
- 重复值情况
2. 所有 KPI、Top N、占比、均值、毛利率等数字，在写入报告前必须再次校验。
3. 如果发现数据不足以支撑某个结论，应降低结论强度，而不是强行给出确定性判断。
保存前请完成自检，确保：
1. 报告中的数据质量信息和关键指标与原始 CSV 及计算结果一致。
2. report.html 不依赖任何外部资源。
3. HTML 内容完整有效，不包含无效字段、占位内容或语法错误。'
再创建 Agent，将上述 System Prompt 与工具一起提交：
agent_payload=$(jq -n \
--arg name "data-analyst-agent" \
--arg model "$MODEL_ID" \
--arg system "$ANALYST_SYSTEM_PROMPT" \
--arg toolset "$TOOLSET_TYPE" \
'{
name: $name,
description: "把 CSV 数据集分析成带交互图表的 HTML 报告。",
model: {id: $model, speed: "standard"},
system: $system,
tools: [{
type: $toolset,
default_config: {
enabled: true,
permission_policy: {type: "always_allow"}
}
}],
skills: [],
mcp_servers: [],
metadata: {scenario: "data_analyst_report"}
}')
agent=$(curl -sS --fail-with-body "$ARK_BASE_URL/agents" \
-H "Authorization: Bearer $ARK_API_KEY" \
-H "Content-Type: application/json" \
-d "$agent_payload")
AGENT_ID=$(jq -er '.id' <<<"$agent")
echo "Agent: $AGENT_ID"
第 3 步：上传待分析数据#
上传输入文件有多种方式，本示例通过 Files API 上传 CSV 并获取 FILE_ID，后续创建 Session 时通过该 ID 引用文件。文件上传采用异步处理，需要等待文件状态变为 active 后再继续后续流程，以确保文件已完成处理并可以正常使用。
file=$(curl -sS --fail-with-body "$ARK_BASE_URL/files" \
-H "Authorization: Bearer $ARK_API_KEY" \
-F 'purpose=agent' \
-F "file=@${DATA_PATH}")
FILE_ID=$(jq -er '.id' <<<"$file")
echo "File: $FILE_ID"
for _ in $(seq 1 60); do
file_info=$(curl -sS --fail-with-body "$ARK_BASE_URL/files/$FILE_ID" \
-H "Authorization: Bearer $ARK_API_KEY")
file_status=$(jq -r '.status // "active"' <<<"$file_info")
[[ "$file_status" == "active" ]] && break
[[ "$file_status" == "failed" ]] && { echo "$file_info" >&2; exit 1; }
sleep 2
done
第 4 步：创建 Session 并挂载数据#
Session 表示一次具体的运行实例，用于关联 Agent、Environment 和输入资源。创建 Session 时，通过 resources 指定需要挂载的文件，平台会将其映射到沙箱中的 /mnt/session/uploads/ 目录，供 Agent 在执行过程中访问。Session 创建完成后即进入就绪状态，但不会自动开始执行任务，需在后续发送事件触发执行。
UPLOAD_NAME=$(basename "$DATA_PATH")
MOUNT_PATH="/mnt/session/uploads/$UPLOAD_NAME"
session_payload=$(jq -n \
--arg agent "$AGENT_ID" \
--arg env "$ENVIRONMENT_ID" \
--arg file "$FILE_ID" \
--arg name "$UPLOAD_NAME" \
'{
agent: $agent,
environment_id: $env,
title: "CSV data analysis",
resources: [{type: "file", file_id: $file, mount_path: $name}]
}')
session=$(curl -sS --fail-with-body "$ARK_BASE_URL/sessions" \
-H "Authorization: Bearer $ARK_API_KEY" \
-H "Content-Type: application/json" \
-d "$session_payload")
SESSION_ID=$(jq -er '.id' <<<"$session")
echo "Session: $SESSION_ID"
第 5 步：建立事件流并下发分析任务#
Session 创建完成后并不会自动执行任务。为了完整接收本轮事件，先建立 SSE 流并等待服务端返回 : ready，再向 /sessions/{session_id}/events 接口发送 user.message。以下函数会按该顺序编排，并持续读取到会话终态：
run_session_turn() (
local session_id="$1"
local user_text="$2"
local stream_dir
local stream_pipe=""
local stream_error
local stream_pid=""
local stream_number=0
local stream_ready=false
local seen_ids_file
local events_payload
local turn_done=false
local result_code=1
local message_count=0
local reconnect_count=0
local max_reconnects="${ARK_MAX_SSE_RECONNECTS:-3}"
local max_history_pages="${ARK_MAX_HISTORY_PAGES:-100}"
local line
local data
: "${ARK_BASE_URL:?Set ARK_BASE_URL before calling run_session_turn}"
: "${ARK_API_KEY:?Set ARK_API_KEY before calling run_session_turn}"
: "${session_id:?Pass a Session ID as the first argument}"
: "${user_text:?Pass a user message as the second argument}"
command -v curl >/dev/null || {
printf 'curl is required.\n' >&2
return 1
}
command -v jq >/dev/null || {
printf 'jq is required.\n' >&2
return 1
}
[[ $max_reconnects =~ ^[0-9]+$ ]] || {
printf 'ARK_MAX_SSE_RECONNECTS must be a non-negative integer.\n' >&2
return 1
}
[[ $max_history_pages =~ ^[1-9][0-9]*$ ]] || {
printf 'ARK_MAX_HISTORY_PAGES must be a positive integer.\n' >&2
return 1
}
stream_dir=$(mktemp -d) || {
printf 'Failed to create a temporary directory for the SSE stream.\n' >&2
return 1
}
stream_error="$stream_dir/stream-error.log"
seen_ids_file="$stream_dir/seen-event-ids"
: >"$stream_error"
: >"$seen_ids_file"
stop_stream() {
exec 3<&- || true
if [[ -n $stream_pid ]] && kill -0 "$stream_pid" 2>/dev/null; then
kill "$stream_pid" 2>/dev/null || true
fi
if [[ -n $stream_pid ]]; then
wait "$stream_pid" 2>/dev/null || true
fi
if [[ -n $stream_pipe ]]; then
rm -f "$stream_pipe"
fi
stream_pid=""
stream_pipe=""
}
cleanup_stream() {
stop_stream
rm -rf "$stream_dir"
}
trap cleanup_stream EXIT
start_stream() {
stop_stream
stream_number=$((stream_number + 1))
stream_pipe="$stream_dir/events-$stream_number"
stream_ready=false
if ! mkfifo "$stream_pipe"; then
printf 'Failed to create the SSE pipe at %s.\n' "$stream_pipe" >&2
return 1
fi
curl -sS -N --fail-with-body --max-time 600 \
"$ARK_BASE_URL/sessions/$session_id/events/stream" \
-H "Authorization: Bearer $ARK_API_KEY" \
-H "Accept: text/event-stream" >"$stream_pipe" 2>>"$stream_error" &
stream_pid=$!
exec 3<"$stream_pipe"
while IFS= read -r line <&3; do
line=${line%$'\r'}
if [[ $line == ": ready" ]]; then
stream_ready=true
break
fi
done
if [[ $stream_ready != true ]]; then
printf 'SSE stream closed before it was ready.\n' >&2
if [[ -s $stream_error ]]; then
tail -n 1 "$stream_error" >&2
fi
return 1
fi
}
event_seen() {
local event_id="$1"
[[ -n $event_id ]] && grep -Fqx -- "$event_id" "$seen_ids_file"
}
remember_event() {
local event_id="$1"
if [[ -n $event_id ]] && ! event_seen "$event_id"; then
printf '%s\n' "$event_id" >>"$seen_ids_file"
fi
}
process_event() {
local event_json="$1"
local event_id
local event_type
local stop_reason
local retry_status
local pending_ids
event_id=$(jq -r '.id // empty' <<<"$event_json" 2>/dev/null || true)
if event_seen "$event_id"; then
return 0
fi
remember_event "$event_id"
event_type=$(jq -r '.type // empty' <<<"$event_json" 2>/dev/null || true)
case "$event_type" in
agent.message)
jq -r '.content[]? | select(.type == "text") | .text' <<<"$event_json"
message_count=$((message_count + 1))
;;
agent.tool_use | agent.mcp_tool_use | agent.custom_tool_use)
printf '[tool] %s\n' "$(jq -r '.name // .type' <<<"$event_json")"
;;
session.error)
retry_status=$(jq -r '.error.retry_status.type // empty' <<<"$event_json")
printf '[error] %s: %s\n' \
"$(jq -r '.error.type // "unknown_error"' <<<"$event_json")" \
"$(jq -r '.error.message // "No error message returned."' <<<"$event_json")" >&2
case "$retry_status" in
retrying)
printf '[session] The service is retrying automatically.\n' >&2
;;
exhausted)
printf '[session] Retries were exhausted for this turn.\n' >&2
;;
terminal)
printf '[session] The Session cannot continue after this error.\n' >&2
;;
*)
printf '[session] Unknown error retry status: %s\n' "$retry_status" >&2
result_code=4
turn_done=true
;;
esac
;;
session.status_rescheduled)
printf '[session] The Session was rescheduled; continue waiting.\n' >&2
;;
session.status_idle)
stop_reason=$(jq -r '.stop_reason.type // empty' <<<"$event_json")
case "$stop_reason" in
end_turn)
if ((message_count == 0)); then
printf 'The turn ended without an agent.message event.\n' >&2
result_code=5
else
printf '[session] end_turn\n' >&2
result_code=0
fi
turn_done=true
;;
requires_action)
pending_ids=$(jq -r '.stop_reason.event_ids // [] | join(",")' <<<"$event_json")
printf '[session] requires_action: %s\n' "$pending_ids" >&2
result_code=2
turn_done=true
;;
retries_exhausted)
printf '[session] retries_exhausted\n' >&2
result_code=3
turn_done=true
;;
*)
printf 'Unknown session.status_idle stop_reason: %s\n' "$stop_reason" >&2
result_code=5
turn_done=true
;;
esac
;;
session.status_terminated)
printf '[session] terminated\n' >&2
result_code=4
turn_done=true
;;
esac
}
read_history() {
local mode="$1"
local page=""
local page_count=0
local response
local event_json
while :; do
page_count=$((page_count + 1))
if ((page_count > max_history_pages)); then
printf 'History recovery exceeded %s pages.\n' "$max_history_pages" >&2
return 1
fi
if [[ -n $page ]]; then
response=$(curl -sS --fail-with-body --get \
"$ARK_BASE_URL/sessions/$session_id/events" \
-H "Authorization: Bearer $ARK_API_KEY" \
--data-urlencode "limit=200" \
--data-urlencode "order=asc" \
--data-urlencode "page=$page") || return 1
else
response=$(curl -sS --fail-with-body --get \
"$ARK_BASE_URL/sessions/$session_id/events" \
-H "Authorization: Bearer $ARK_API_KEY" \
--data-urlencode "limit=200" \
--data-urlencode "order=asc") || return 1
fi
while IFS= read -r event_json; do
[[ -n $event_json ]] || continue
if [[ $mode == "seed" ]]; then
remember_event "$(jq -r '.id // empty' <<<"$event_json")"
else
process_event "$event_json"
[[ $turn_done == true ]] && break
fi
done < <(jq -c '.data[]?' <<<"$response")
[[ $turn_done == true ]] && return 0
page=$(jq -r '.next_page // empty' <<<"$response")
[[ -n $page ]] || return 0
done
}
if ! start_stream; then
return 1
fi
# Mark existing history so reconnect recovery only emits this turn.
if ! read_history seed; then
printf 'Failed to read the Session history before sending the message.\n' >&2
return 1
fi
events_payload=$(jq -n --arg text "$user_text" \
'{events: [{type: "user.message", content: [{type: "text", text: $text}]}]}')
curl -sS --fail-with-body \
"$ARK_BASE_URL/sessions/$session_id/events" \
-H "Authorization: Bearer $ARK_API_KEY" \
-H "Content-Type: application/json" \
-d "$events_payload" >/dev/null || return 1
while [[ $turn_done != true ]]; do
while IFS= read -r line <&3; do
line=${line%$'\r'}
if [[ $line == ": server draining, reconnect" ]]; then
break
fi
[[ $line == data:* ]] || continue
data=${line#data:}
data=${data# }
process_event "$data"
[[ $turn_done == true ]] && break
done
[[ $turn_done == true ]] && break
if ((reconnect_count >= max_reconnects)); then
printf 'SSE reconnect limit reached before the turn finished.\n' >&2
return 1
fi
reconnect_count=$((reconnect_count + 1))
printf '[session] Reconnecting SSE and recovering history (%s/%s).\n' \
"$reconnect_count" "$max_reconnects" >&2
# Open the replacement stream first, then recover the disconnect window.
if ! start_stream; then
return 1
fi
if ! read_history recover; then
printf 'Failed to recover Session history after reconnecting.\n' >&2
return 1
fi
done
return "$result_code"
)
analysis_prompt=$(cat <<EOF
分析挂载在 $MOUNT_PATH 的 CSV 数据集。
请完成：
1. 识别字段含义、数据规模和明显质量问题。
2. 找出收入、品类、区域、时间趋势或用户行为中的关键发现。
3. 至少生成 3 张 plotly 交互图表。
4. 输出一个 HTML 报告，包含摘要、图表、关键数字和行动建议。
EOF
)
run_session_turn "$SESSION_ID" "$analysis_prompt"
第 6 步：实时观测执行过程#
第 5 步建立的 SSE 流会实时输出模型消息、工具调用和任务状态。run_session_turn 只有在收到 session.status_idle 且 stop_reason.type=end_turn，并且本轮至少产生一条 agent.message 时才返回成功。
如果 SSE 意外断开，函数会先建立替代流，再分页查询历史事件，补齐断线期间的事件并按事件 ID 去重。连续重连仍失败、Session 终止或服务端返回其他停止原因时，函数会返回非零状态，不会把不完整输出当作成功结果。
主要事件类型：
| 事件类型 | 含义 |
|---|---|
agent.message |
Agent 返回的文本消息 |
agent.tool_use |
Agent 正在调用工具 |
agent.tool_result |
工具执行结果 |
session.status_idle |
根据 stop_reason.type 判断本轮正常结束、等待操作或重试耗尽 |
session.status_terminated |
Session 已终止，需要排查原因 |
第 7 步：取回分析报告#
任务完成后，可以通过 Session ID（scope_id）查询本次执行生成的文件。Agent 运行过程中产生的报告、图表等文件都会关联到对应的 Session。当查询结果中出现生成的 HTML 文件时，说明分析报告已经生成。
curl -sS --fail-with-body "$ARK_BASE_URL/files?scope_id=$SESSION_ID" \
-H "Authorization: Bearer $ARK_API_KEY" |
jq '.data[]? | {id, filename, bytes, status}'
生成的文件可以通过控制台下载到本地，方便查看、分享或归档。
第 8 步：清理资源#
分析完成后，可根据需要清理本次运行产生的资源。Session 和上传文件属于临时资源，会占用平台配额，确认结果已保存后可及时删除。Agent 和 Environment 通常作为可复用资源保留，用于后续任务。再次执行分析时，只需创建新的 Session 并引用已有的 Agent、Environment 和输入文件即可。
curl -sS --fail-with-body -X DELETE \
"$ARK_BASE_URL/sessions/$SESSION_ID" \
-H "Authorization: Bearer $ARK_API_KEY"
curl -sS --fail-with-body -X DELETE \
"$ARK_BASE_URL/files/$FILE_ID" \
-H "Authorization: Bearer $ARK_API_KEY"
curl -sS --fail-with-body -X DELETE \
"$ARK_BASE_URL/environments/$ENVIRONMENT_ID" \
-H "Authorization: Bearer $ARK_API_KEY"
完整执行顺序#
- 设置
ARK_API_KEY、ARK_BASE_URL、MODEL_ID、DATA_PATH等环境变量。 - 创建或复用 Environment，确保包含
pandas和plotly。 - 创建 Agent。
- 上传 CSV。
- 创建 Session 并挂载 CSV。
- 建立事件流并等待
: ready。 - 发送
user.message，通过同一条事件流等待session.status_idle且stop_reason.type=end_turn。 - 按
scope_id查询 Session 产物。 - 下载生成文件到本地。
- 按需清理 Session、输入文件与 Environment。