Agent 工程体系 · 第 60/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

数据分析 Agent：文件、代码执行、表格、图表、复现和结果验证

数据分析 Agent 不是“会写 Python 的聊天机器人”。它是一个能够接收数据文件、理解数据结构、生成并执行分析程序、产出表格与图表、保存运行证据，并对最终结果进行独立校验的系统。

如果只让模型生成一段代码，系统最多是代码生成器；如果只让模型返回一组数字，系统无法回答“这些数字来自哪个文件、经过了哪些过滤、能否重新得到、是否存在口径错误”。真正的数据分析 Agent 必须把数据、代码、执行环境、产物、结论和验证证据绑定在同一条可追踪链路上。

本文讨论的核心问题是：

$$\text{用户问题} \rightarrow \text{数据发现} \rightarrow \text{分析计划} \rightarrow \text{代码执行} \rightarrow \text{结果产出} \rightarrow \text{结果验证} \rightarrow \text{可复现报告}$$

其中任何一个环节缺失，系统都可能产生“看起来合理、实际上无法证明”的答案。

---

一、先定义对象：数据分析 Agent 到底负责什么

1. 文件不是数据，文件是数据的载体

常见文件包括：

• CSV、TSV；
• Excel；
• JSON、JSON Lines；
• Parquet；
• 数据库导出的 SQL 结果；
• 图片、PDF 中的表格；
• 压缩包；
• 由上游任务生成的中间文件。

文件本身至少包含四类信息：

1. 字节内容：文件实际存储的二进制数据；
2. 格式信息：编码、分隔符、工作表、压缩格式、列类型；
3. 业务元数据：来源、生成时间、数据所有者、时间范围；
4. 访问上下文：谁上传、谁有权限、是否允许写回或外发。

因此，Agent 不应仅把文件名放进提示词，例如：

请分析 sales.xlsx

更可靠的内部对象应类似于：

{
  "artifact_id": "file_01JABC...",
  "original_name": "sales.xlsx",
  "sha256": "9d8c...",
  "size_bytes": 184293,
  "media_type": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
  "created_at": "2026-09-01T08:30:00Z",
  "source": "user_upload",
  "access_scope": "workspace:finance",
  "read_only": true
}

artifact_id 是系统内部稳定引用，sha256 用于确认内容是否改变。分析过程引用的是 artifact_id 和内容摘要，而不是依赖用户提供的文件名。

2. 数据分析 Agent 的输出不是一句话

一个完整的分析输出至少包含：

$$O = (D, P, C, E, A, V, R)$$

其中：

• $D$：输入数据快照（Data snapshot）；
• $P$：分析计划（Plan）；
• $C$：实际执行的代码（Code）；
• $E$：执行环境（Environment）；
• $A$：分析产物（Artifacts）；
• $V$：验证结果（Validation）；
• $R$：面向用户的报告（Report）。

例如，报告中的“华东区域收入同比增长 12.4%”不能只保存这句话，还应能够追溯到：

输入文件：file_01JABC...
内容哈希：9d8c...
过滤条件：region == "华东"
时间字段：order_date
基准期：2025-01-01 至 2025-06-30
比较期：2026-01-01 至 2026-06-30
聚合公式：sum(amount)
实际代码：run_7f31...
输出表格：artifact_table_03...
验证状态：passed

这套关系可以称为证据图：

flowchart LR
    Q[用户问题] --> P[分析计划]
    F[文件快照] --> P
    S[Schema 与数据剖面] --> P
    P --> C[分析代码]
    C --> E[隔离执行环境]
    E --> T[表格产物]
    E --> G[图表产物]
    E --> L[日志与指标]
    T --> V[结果验证]
    G --> V
    L --> V
    V --> R[最终报告]
    C --> X[复现包]
    F --> X
    E --> X

报告中的结论必须能够沿图中的边反向追踪。无法追踪的结论只能标记为“未经验证的模型解释”，不应伪装成计算结果。

---

二、数据分析 Agent 的基本架构

数据分析 Agent 通常由以下组件组成：

1. 文件管理器：接收、识别、校验和读取文件；
2. 数据剖面器：统计列名、类型、缺失值、唯一值、异常值和样本；
3. 分析规划器：把自然语言问题转换为可执行步骤；
4. 代码生成器：生成 SQL、Python、R 或其他分析代码；
5. 代码执行器：在隔离环境中运行代码；
6. 产物管理器：保存表格、图表、日志和中间结果；
7. 验证器：检查数据、计算、图表和结论；
8. 报告生成器：把已验证结果转换为用户可读内容；
9. 审计与复现存储：保存输入、代码、环境、日志和版本。

模型不应直接拥有主机文件系统或数据库连接。模型只能提出工具调用，宿主程序负责校验、授权和执行。OpenAI 的 Function Calling 机制也是这种模式：模型返回工具调用，应用侧执行工具，再把工具输出连同调用标识返回给模型，最终模型才生成回答。官方文档将其描述为“请求模型—接收工具调用—应用侧执行—提交工具输出—获得最终响应”的多步流程。(developers.openai.com)

一个最小的控制流如下：

sequenceDiagram
    participant U as 用户
    participant A as Agent
    participant H as 宿主编排器
    participant S as 沙箱
    participant V as 验证器
    participant R as 产物存储

    U->>A: 分析请求
    A->>H: 请求文件元数据/数据剖面
    H-->>A: Schema、统计摘要、样本
    A->>H: 提交分析计划
    H->>H: 校验计划与权限
    H->>S: 创建一次性运行环境
    A->>H: 提交代码
    H->>S: 执行代码
    S-->>H: stdout、stderr、产物清单、退出码
    H->>V: 独立验证结果
    V-->>H: 通过/失败及证据
    H->>R: 保存代码、日志、表格、图表、验证报告
    H-->>A: 返回已验证产物
    A-->>U: 解释、表格、图表和复现信息

这里的关键不是“模型能不能调用工具”，而是宿主编排器是否在工具调用前后承担了类型校验、权限控制、资源限制和证据保存。

---

三、文件处理：从上传字节到可分析数据

3.1 文件接入必须先建立不变量

文件进入系统后，至少应建立以下不变量：

$$I_f = \begin{cases} \text{content\_hash is stable} \\ \text{media\_type is detected, not merely trusted} \\ \text{size} \leq \text{configured limit} \\ \text{reader matches detected format} \\ \text{access scope is known} \end{cases}$$

直觉上，这些条件回答了五个问题：

• 分析的文件内容有没有被替换？
• 文件扩展名是否骗人？
• 是否可能造成内存或磁盘耗尽？
• 是否使用了正确的解析器？
• 该文件是否属于当前用户和当前任务？

不能只依赖扩展名判断格式：

from pathlib import Path
import hashlib

def inspect_file(path: str) -> dict:
    p = Path(path)
    data = p.read_bytes()

    return {
        "name": p.name,
        "suffix": p.suffix.lower(),
        "size_bytes": len(data),
        "sha256": hashlib.sha256(data).hexdigest(),
        "prefix": data[:16].hex(),
    }

print(inspect_file("sales.csv"))

prefix 可以辅助识别文件头，但生产系统应使用成熟的 MIME 检测和格式解析器，并对压缩包展开后的总大小、文件数量和路径穿越进行限制。

3.2 先做数据剖面，再生成分析代码

数据剖面（data profiling）是对数据集结构和基本统计特征的机器化描述。一个最小剖面包括：

• 行数、列数；
• 列名；
• 推断类型与实际解析类型；
• 缺失数量和比例；
• 唯一值数量；
• 最小值、最大值、分位数；
• 重复行数量；
• 时间范围；
• 示例值；
• 解析警告。

示例：

import pandas as pd

df = pd.read_csv("sales.csv")

profile = pd.DataFrame({
    "dtype": df.dtypes.astype(str),
    "missing": df.isna().sum(),
    "missing_ratio": df.isna().mean(),
    "nunique": df.nunique(dropna=True),
})

print("shape =", df.shape)
print(profile)
print(df.head(3))

如果 amount 被解析成字符串，Agent 不能直接生成：

df["amount"].sum()

因为结果可能是字符串拼接、异常或静默转换。正确做法是先明确转换策略：

df["amount"] = (
    df["amount"]
      .astype("string")
      .str.replace(",", "", regex=False)
      .str.strip()
)

df["amount"] = pd.to_numeric(df["amount"], errors="coerce")

随后必须报告被转换为缺失值的数量：

coercion_count = df["amount"].isna().sum()
print("amount coercion/missing:", coercion_count)

这一步很重要：类型转换不是清洗细节，而是计算语义的一部分。如果 1000 条金额中有 37 条无法解析，最终总额是否包含这 37 条，必须成为可见的决策。

3.3 文件读取的常见边界

编码问题

同一个 CSV 可能使用 UTF-8、GBK 或带 BOM 的 UTF-8。读取失败时，系统不能让模型“猜一个编码然后继续”。应记录：

尝试编码：utf-8
结果：UnicodeDecodeError
尝试编码：utf-8-sig
结果：成功

Excel 工作表问题

Excel 文件可能有多个工作表、隐藏表、合并单元格和公式。分析器应先列出工作表：

import pandas as pd

book = pd.ExcelFile("sales.xlsx")
print(book.sheet_names)

模型选择工作表后，宿主程序应把选择记录到计划中：

{
  "file": "file_01JABC",
  "sheet": "2026_H1",
  "header_row": 2,
  "reason": "该工作表包含 order_date、region、amount 三列"
}

大文件问题

大文件不能无条件 read_csv 到内存。应使用分块读取：

import pandas as pd

total = 0.0
rows = 0

for chunk in pd.read_csv("large_sales.csv", chunksize=100_000):
    amount = pd.to_numeric(chunk["amount"], errors="coerce")
    total += amount.sum()
    rows += len(chunk)

print({"rows": rows, "amount_sum": total})

但分块聚合只适用于满足可结合性的操作。求和、计数、最小值可以流式计算；精确中位数、全局排序和复杂窗口函数通常需要额外算法或外部存储。不能因为“代码能跑”就认为计算语义正确。

---

四、代码执行：模型生成代码，不等于模型拥有执行权

4.1 代码执行器的职责

代码执行器不是简单的：

exec(model_generated_code)

它至少要负责：

• 创建独立工作目录；
• 注入只读输入文件；
• 限制 CPU、内存、磁盘和运行时间；
• 限制系统调用；
• 控制网络访问；
• 固定或声明依赖版本；
• 捕获标准输出和错误输出；
• 收集退出码和信号；
• 保存生成的文件；
• 超时或异常时终止全部子进程；
• 任务结束后销毁环境。

可以把一次执行表示为：

$$X = \operatorname{Run}(C, D, E, L)$$

其中：

• $C$：代码；
• $D$：输入数据快照；
• $E$：执行环境；
• $L$：资源限制；
• $X$：包含退出码、输出、错误和产物的执行结果。

如果 $E$ 未记录，代码即使保存了，也不一定能复现。比如同一段代码在不同版本的 pandas 中可能产生不同的类型推断或排序行为。

4.2 进程、容器和解释器隔离的区别

仅使用独立进程

优点是启动快、实现简单；缺点是 Python 进程仍可能访问宿主文件系统、环境变量和网络。它适合可信代码，不适合作为通用 Agent 沙箱的唯一边界。

使用容器

容器可以提供文件系统、用户、网络和资源隔离，但容器配置本身不是安全保证。错误的特权配置、挂载宿主目录、暴露 Docker socket，都可能破坏隔离。

使用更强的沙箱或微虚拟机

适合运行不可信代码，但启动、调度和镜像管理成本更高。生产选择取决于威胁模型：内部可信用户、跨租户 SaaS、处理敏感数据，所需边界并不相同。

代码执行安全性不应由模型自觉保证。模型可能生成：

import os
print(os.environ)

或：

import socket
# 尝试访问外部服务

因此，禁止网络访问、清理环境变量和限制挂载点必须由执行层实现，而不是写在 system prompt 中。

4.3 沙箱生命周期

一个可审计的执行生命周期可以定义为：

CREATED
  -> INPUT_ATTACHED
  -> CODE_VALIDATED
  -> RUNNING
  -> SUCCEEDED
  -> ARTIFACTS_COLLECTED
  -> VERIFIED
  -> DESTROYED

异常路径则可能是：

RUNNING
  -> TIMED_OUT
  -> PROCESS_GROUP_KILLED
  -> ARTIFACTS_QUARANTINED
  -> DESTROYED

这里的 PROCESS_GROUP_KILLED 不能省略。只杀父进程可能留下子进程继续消耗资源，甚至继续写入文件或访问网络。

一个简化的本地执行示例：

import os
import signal
import subprocess
import tempfile
from pathlib import Path

def run_code(code: str, timeout_seconds: int = 10) -> dict:
    with tempfile.TemporaryDirectory(prefix="analysis-") as tmp:
        workdir = Path(tmp)
        script = workdir / "analysis.py"
        script.write_text(code, encoding="utf-8")

        env = {
            "PATH": os.environ.get("PATH", ""),
            "PYTHONUNBUFFERED": "1",
        }

        process = subprocess.Popen(
            ["python", str(script)],
            cwd=workdir,
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            start_new_session=True,
        )

        try:
            stdout, stderr = process.communicate(timeout=timeout_seconds)
        except subprocess.TimeoutExpired:
            os.killpg(process.pid, signal.SIGKILL)
            stdout, stderr = process.communicate()
            return {
                "status": "timed_out",
                "stdout": stdout,
                "stderr": stderr,
                "returncode": process.returncode,
            }

        return {
            "status": "succeeded" if process.returncode == 0 else "failed",
            "stdout": stdout,
            "stderr": stderr,
            "returncode": process.returncode,
            "artifacts": [p.name for p in workdir.iterdir()],
        }

这个示例只说明生命周期和进程组处理，不能直接当作生产安全沙箱。它仍然需要操作系统级资源限制、网络隔离、系统调用策略、输入挂载策略和依赖供应链控制。

---

五、表格：不是把 DataFrame 打印出来

5.1 表格是可检查的结果数据结构

表格产物应区别于终端文本。至少要有：

{
  "artifact_id": "table_01",
  "kind": "table",
  "columns": [
    {"name": "region", "type": "string"},
    {"name": "revenue", "type": "number"},
    {"name": "order_count", "type": "integer"}
  ],
  "row_count": 3,
  "storage": "parquet",
  "preview": [
    {"region": "华东", "revenue": 120000, "order_count": 310}
  ]
}

表格的三个层次应分开：

1. 原始表：输入文件读取后的数据；
2. 中间表：过滤、连接、分组后的数据；
3. 展示表：为报告格式化后的数据。

例如金额保留两位小数适合展示，但不应覆盖原始数值。否则后续计算可能因为重复四舍五入产生累计误差。

5.2 聚合必须明确粒度

考虑订单表：

order_id	region	amount
1	华东	100
2	华东	200
3	华南	50

“华东收入”通常表示：

$$R_{\text{华东}} = \sum_{i:\ region_i=\text{华东}} amount_i = 100+200=300$$

但如果一张订单表被连接到商品明细表：

order_id	sku	quantity
1	A	1
1	B	2
2	C	1

直接连接后，订单 1 会出现两行。如果再对订单金额求和：

$$100 + 100 + 200 = 400$$

这就是典型的连接放大。正确做法是先确认分析粒度：

orders = pd.DataFrame({
    "order_id": [1, 2, 3],
    "region": ["华东", "华东", "华南"],
    "amount": [100, 200, 50],
})

items = pd.DataFrame({
    "order_id": [1, 1, 2],
    "sku": ["A", "B", "C"],
    "quantity": [1, 2, 1],
})

item_summary = items.groupby("order_id", as_index=False)["quantity"].sum()

result = (
    orders
    .merge(item_summary, on="order_id", how="left", validate="one_to_one")
    .groupby("region", as_index=False)
    .agg(
        revenue=("amount", "sum"),
        order_count=("order_id", "nunique"),
        item_quantity=("quantity", "sum"),
    )
)

validate="one_to_one" 不是装饰，它把“订单表与商品汇总表是一对一关系”转成了可执行约束。如果数据不满足，程序应失败，而不是继续输出一个看似正常的数字。

---

六、图表：图形编码的是变量关系，不是装饰

6.1 图表必须有数据契约

一张图表至少应记录：

{
  "artifact_id": "chart_01",
  "chart_type": "line",
  "x": "month",
  "y": "revenue",
  "series": "region",
  "aggregation": "sum(amount)",
  "filters": ["status == 'paid'"],
  "source_table": "table_01",
  "format": "png"
}

这说明图表不是一张孤立的 PNG，而是一个可解释的可视化产物。

图表选择取决于变量类型和分析目的：

• 时间趋势：折线图；
• 类别比较：排序后的条形图；
• 分布：直方图、箱线图；
• 两个数值变量的关系：散点图；
• 组成结构：堆叠条形图，但类别过多时不适合；
• 空间关系：地图，但必须明确地理边界和投影。

6.2 图表常见的统计误导

坐标轴截断

柱状图的长度通常被理解为从零开始的数量编码。如果纵轴从 99 开始，100 与 101 的差异会被视觉放大。折线图有时可以截断坐标轴，但必须明确标注。

汇总掩盖分布

平均值相同不代表数据相同：

样本 A: 49, 50, 51
样本 B: 0, 50, 100

两者平均值都是 50，但方差、风险和业务含义不同。Agent 不能只画均值趋势而声称“表现稳定”，除非同时检查离散程度。

排序和类别顺序

类别条形图若按输入顺序排列，用户可能误以为是排名。若需要排名，必须明确排序规则：

summary = (
    df.groupby("region", as_index=False)
      .agg(revenue=("amount", "sum"))
      .sort_values("revenue", ascending=False)
)

时间聚合的缺失月份

如果某月份没有记录，直接画线可能把“没有数据”误认为“收入为零”。应构造完整时间索引，再区分：

• 没有观测；
• 观测值为零；
• 数据尚未到达。

---

七、复现：能重新运行，不等于能重新得到

7.1 复现的三种强度

代码复现

保存代码，重新执行即可。这是最低层次，但依赖输入文件和环境没有变化。

结果复现

在相同输入和环境下，重新得到相同结果。可以定义：

$$\operatorname{Digest}(A_{\text{new}}) = \operatorname{Digest}(A_{\text{old}})$$

其中 $A$ 是表格或结构化结果。对于浮点数、排序和随机采样，需要定义容差与排序规则。

解释复现

不仅数字相同，用户还能知道结果使用了哪些字段、过滤条件、公式和假设。这是面向业务审计最重要的层次。

7.2 复现包的最小组成

一个复现包可以包含：

run_7f31/
├── manifest.json
├── input/
│   └── file_01JABC.sha256
├── code/
│   ├── profile.py
│   ├── analysis.py
│   └── verify.py
├── environment/
│   ├── requirements.lock
│   └── runtime.json
├── output/
│   ├── summary.parquet
│   └── revenue_by_region.png
├── logs/
│   ├── stdout.log
│   └── stderr.log
└── validation/
    └── report.json

manifest.json 可记录：

{
  "run_id": "run_7f31",
  "input_hashes": {
    "sales.csv": "9d8c..."
  },
  "code_hash": "1aa2...",
  "runtime": {
    "python": "3.12.x",
    "timezone": "Asia/Shanghai"
  },
  "random_seed": 20260901,
  "status": "verified",
  "created_at": "2026-09-01T10:20:00+08:00"
}

随机性必须显式处理。若代码使用抽样、随机初始化或机器学习模型，应记录随机种子；但仅记录种子不一定足够，因为不同库版本、不同并行调度和不同硬件仍可能改变结果。

7.3 时间和时区是复现陷阱

“2026-09-01 的数据”必须说明时区和边界：

[2026-09-01T00:00:00+08:00,
 2026-09-02T00:00:00+08:00)

相比 <= 2026-09-01 23:59:59，半开区间更不容易遗漏带毫秒的数据。

---

八、结果验证：验证数字，不验证模型的自信

结果验证的目标不是再次问模型“你确定吗”，而是使用可执行的独立条件判断结果是否可信。

8.1 验证的四个层次

1. 语法和运行验证

检查：

• 代码是否成功退出；
• 是否存在未捕获异常；
• 输出文件是否存在；
• 输出文件是否可读；
• 日志中是否有警告。

2. 数据质量验证

检查：

• 主键是否唯一；
• 必填字段是否为空；
• 金额是否出现负值；
• 日期是否超出预期；
• 类别值是否在允许集合中；
• 连接前后行数是否符合预期。

3. 计算不变量验证

不变量是无论实现细节如何都应成立的关系。

订单统计示例：

$$\sum_{r \in Regions} Revenue_r = Revenue_{\text{total}}$$

代码：

region_total = result["revenue"].sum()
raw_total = orders["amount"].sum()

assert abs(region_total - raw_total) < 1e-9

分类计数示例：

$$\sum_{c \in Categories} Count_c = Count_{\text{valid rows}}$$

如果存在“其他”“未知”或缺失类别，必须把它们纳入分区，否则等式不成立。

4. 独立实现验证

最强的验证方式之一，是用不同路径重新计算同一个结果。例如主分析使用 pandas，验证器使用 SQL 或纯 Python：

from collections import defaultdict

expected = defaultdict(float)

for row in orders.itertuples(index=False):
    expected[row.region] += row.amount

expected = dict(expected)

actual = dict(zip(result["region"], result["revenue"]))

for region, value in expected.items():
    assert abs(actual[region] - value) < 1e-9

主代码如果错误地使用了重复连接，独立实现可能仍然使用同样错误的数据中间表。因此，验证器不仅要“换一种语法”，还要尽量使用不同的中间路径和不同的假设。

8.2 一个完整算例：区域收入分析

输入：

order_id	order_date	region	status	amount
1	2026-01-03	华东	paid	100
2	2026-01-05	华东	cancelled	200
3	2026-01-08	华南	paid	50
4	2026-02-02	华东	paid	80

用户问题：

统计 2026 年 1 月已支付订单的区域收入和订单数。

分析条件：

$$2026\text{-}01\text{-}01 \leq order\_date < 2026\text{-}02\text{-}01$$

$$status = paid$$

因此有效行是订单 1 和订单 3：

order_id	region	amount
1	华东	100
3	华南	50

聚合结果：

region	revenue	order_count
华东	100	1
华南	50	1

完整代码：

import pandas as pd

df = pd.DataFrame({
    "order_id": [1, 2, 3, 4],
    "order_date": ["2026-01-03", "2026-01-05", "2026-01-08", "2026-02-02"],
    "region": ["华东", "华东", "华南", "华东"],
    "status": ["paid", "cancelled", "paid", "paid"],
    "amount": [100, 200, 50, 80],
})

df["order_date"] = pd.to_datetime(df["order_date"])

start = pd.Timestamp("2026-01-01")
end = pd.Timestamp("2026-02-01")

filtered = df[
    (df["order_date"] >= start)
    & (df["order_date"] < end)
    & (df["status"] == "paid")
].copy()

result = (
    filtered
    .groupby("region", as_index=False)
    .agg(
        revenue=("amount", "sum"),
        order_count=("order_id", "nunique"),
    )
    .sort_values("region")
)

assert filtered["order_id"].nunique() == result["order_count"].sum()
assert result["revenue"].sum() == filtered["amount"].sum()

print(result.to_string(index=False))

预期输出：

region  revenue  order_count
    华东      100            1
    华南       50            1

这里的两个断言分别验证：

1. 区域订单数之和等于有效订单数；
2. 区域收入之和等于过滤后金额总和。

如果 Agent 错把 cancelled 也纳入，第二个结果可能变成华东 300；如果 Agent 把 2 月订单纳入，华东会变成 180。业务结论虽然仍然“看起来合理”，但不变量和时间条件能够暴露错误。

8.3 验证失败后的处理

验证失败不是“忽略警告后继续回答”。系统应根据失败类型处理：

失败类型	处理方式
代码语法错误	返回错误给 Agent，允许修复后重跑
类型解析异常	请求澄清或采用显式转换并报告影响
主键重复	停止聚合，要求确认去重规则
汇总不守恒	标记结果无效，检查过滤、连接和缺失类别
图表与表格不一致	丢弃图表，重新从已验证表格生成
超时或资源耗尽	终止任务，缩小范围或改用分块/SQL
输入文件变化	固定旧快照，不允许静默使用新文件

只有通过验证的结果才能进入最终报告。无法验证的结果可以展示，但必须明确标记其状态。

---

九、Function Calling 和 MCP 如何接入数据分析 Agent

9.1 Function Calling：把动作变成受约束的接口

分析 Agent 的工具不应设计成一个万能接口：

{
  "name": "run_any_code",
  "parameters": {
    "code": "string"
  }
}

它虽然灵活，但权限边界、输入语义和审计粒度都很差。更好的方式是把工具拆成有明确职责的接口：

[
  {
    "type": "function",
    "name": "inspect_file",
    "description": "读取文件元数据和数据剖面，不修改输入文件",
    "strict": true,
    "parameters": {
      "type": "object",
      "properties": {
        "artifact_id": {"type": "string"},
        "sample_rows": {"type": ["integer", "null"]}
      },
      "required": ["artifact_id", "sample_rows"],
      "additionalProperties": false
    }
  },
  {
    "type": "function",
    "name": "execute_analysis",
    "description": "在一次性隔离环境中执行分析代码",
    "strict": true,
    "parameters": {
      "type": "object",
      "properties": {
        "input_artifact_ids": {
          "type": "array",
          "items": {"type": "string"}
        },
        "code": {"type": "string"},
        "timeout_seconds": {"type": "integer"}
      },
      "required": ["input_artifact_ids", "code", "timeout_seconds"],
      "additionalProperties": false
    }
  }
]

OpenAI 的 Function Calling 文档说明，启用 strict mode 后，参数对象需要关闭额外属性，并将属性标记为 required；可选字段可以用包含 null 的类型表达。strict mode 的作用是让工具调用更可靠地符合声明的 schema，而不是只做尽力匹配。(developers.openai.com)

但 schema 校验不等于业务授权。即使参数符合 JSON Schema，宿主程序仍要检查：

artifact_id 是否属于当前任务
timeout_seconds 是否超过上限
代码是否试图写入输入目录
工具是否允许网络
工具调用者是否有权限访问该文件

如果多个工具调用互不依赖，可以并发执行；例如同时读取两个文件的剖面。但“先生成清洗表、再读取清洗表”存在数据依赖，不能并发。OpenAI 文档支持模型在某些条件下发起并行函数调用，也允许通过 parallel_tool_calls: false 禁止一个回合产生多个工具调用。(developers.openai.com)

9.2 MCP：标准化上下文、工具和资源

MCP 是连接 LLM 应用与外部数据源和工具的开放协议。当前官方规范页面对应的版本路径为 2026-07-28；规范使用 JSON-RPC 2.0 消息，并区分 Host、Client 和 Server。(modelcontextprotocol.io)

在数据分析场景中，可以这样映射：

MCP 概念	数据分析对象
Resource	文件快照、Schema、数据剖面、已验证表格
Tool	读取文件、执行 SQL、运行代码、生成图表、校验结果
Prompt	标准化分析模板、财务口径、实验分析流程
Progress	大文件读取、长时间运行、分块聚合进度
Cancellation	用户取消分析或超时终止
Error reporting	解析错误、权限错误、执行错误、验证错误

MCP 规范将 Resources 定义为供用户或模型使用的上下文和数据，将 Tools 定义为供 AI 系统执行的函数；同时还定义了进度跟踪、取消和错误报告等通用能力。(modelcontextprotocol.io)

但 MCP 只标准化连接方式，不会自动让工具安全。规范明确提醒，工具可能代表任意代码执行能力，工具描述和注释不能天然被视为可信；宿主应在调用工具前获得明确用户同意，并让用户理解工具行为。(modelcontextprotocol.io)

因此，MCP Server 不应把“任意执行 SQL”“任意读取路径”“任意访问网络”直接暴露给模型。应在 Server 或更下层实现：

• 只读数据库账户；
• 允许访问的 schema 和表；
• SQL 语法与语句类型校验；
• 查询成本和扫描量限制；
• 文件资源范围；
• 脱敏和行级权限；
• 工具调用审计。

---

十、SQL Agent 与代码分析 Agent 的边界

SQL Agent 适合：

• 结构化数据库；
• 大规模聚合；
• 过滤、连接和窗口计算；
• 需要数据库索引和权限治理的场景。

代码执行 Agent 适合：

• 文件格式复杂；
• 统计建模；
• 自定义清洗；
• 图表和报告生成；
• 需要多个 Python 库协作的场景。

二者不应互相替代。一个常见架构是：

数据库 -> SQL Agent 生成只读查询 -> 查询结果快照
                                      |
                                      v
                              Python Agent 做统计和图表

SQL Agent 生成的查询也必须进入验证流程：

1. 解析 SQL；
2. 拒绝 INSERT、UPDATE、DELETE、DROP 等写操作；
3. 校验表和列是否在允许范围；
4. 检查是否缺少时间过滤；
5. 估算扫描量；
6. 执行 EXPLAIN 或等价计划检查；
7. 设置超时和返回行数上限；
8. 保存 SQL、参数、执行用户和数据库快照信息。

尤其要区分：

SELECT SUM(amount) FROM orders;

和：

SELECT SUM(amount)
FROM orders
WHERE status = 'paid'
  AND order_date >= '2026-01-01'
  AND order_date < '2026-02-01';

前者可能在语法上正确，但在业务问题上不完整。查询校验不仅是 SQL 语法校验，更是问题条件是否被落实到查询中的语义校验。

---

十一、状态、并发和故障路径

11.1 Agent 状态不能只保存一段对话

建议把一次分析建模为显式状态：

{
  "run_id": "run_7f31",
  "status": "VERIFYING",
  "question": "统计2026年1月已支付订单的区域收入",
  "input_artifacts": ["file_01JABC"],
  "plan_version": 2,
  "code_attempt": 2,
  "execution_id": "exec_91a2",
  "output_artifacts": ["table_01", "chart_01"],
  "validation": {
    "row_count_check": "passed",
    "sum_conservation": "passed",
    "independent_recompute": "passed"
  }
}

状态机至少要防止以下错误：

• 代码执行失败却进入报告生成；
• 验证旧结果时引用了新文件；
• 用户重试导致同一个副作用工具执行两次；
• 并发任务互相覆盖临时文件；
• 超时任务仍持有数据库连接；
• 图表生成成功但对应表格尚未验证。

11.2 并发的核心是依赖关系

将步骤表示成有向无环图：

inspect(file_a) ─┐
                  ├─> plan ─> execute ─> validate ─> report
inspect(file_b) ─┘

两个 inspect 可以并发，因为它们只读且互不依赖。execute 必须等待 plan，report 必须等待 validate。

并发执行还要求：

• 每个任务拥有独立工作目录；
• 产物 ID 全局唯一；
• 写入操作幂等；
• 取消信号能够传播到所有子任务；
• 失败任务不会被误标为成功；
• 结果合并有明确的版本和顺序。

11.3 故障不是一个“工具报错”

故障应按层区分：

文件层：无法读取、编码错误、格式错误
数据层：列缺失、类型错误、主键重复
计划层：问题含义不明确、口径冲突
代码层：语法错误、导入错误、逻辑错误
执行层：超时、内存不足、网络拒绝
产物层：文件缺失、格式损坏、图表为空
验证层：不变量失败、独立计算不一致
解释层：数字正确但结论越过证据

每一层的恢复策略不同。语法错误通常可以自动修复；口径冲突需要询问用户；主键重复不能由模型擅自决定去重规则；验证失败必须阻止错误结果流入报告。

---

十二、常见误解与反例

误解一：模型执行了代码，所以结果一定正确

反例：

df.groupby("region")["amount"].mean()

如果用户问的是收入总额，代码成功运行并不代表语义正确。执行成功只证明：

$$\text{Syntax Correct} \land \text{Runtime Completed}$$

不代表：

$$\text{Business Meaning Correct}$$

误解二：表格和图表是同一个结果的两种展示

反例：表格按“支付日期”聚合，图表却按“下单日期”聚合。两者都可以单独生成，视觉上也可能接近，但它们不是同一个指标。

因此图表必须引用具体的已验证表格或查询结果，而不是重新让模型生成一套相似代码。

误解三：保存代码就能复现

反例：

df.sample(100)

没有随机种子时，每次样本都可能不同。即使加了种子，如果输入数据、库版本、时区或排序规则变化，结果仍可能变化。

误解四：数据量少就不需要验证

小数据更容易被人工“看起来正常”误导。最危险的错误往往不是程序崩溃，而是少算一个月份、重复连接一张表或误把空值当零。数据量越小，越应该用完整算例和守恒关系检查。

误解五：只读 SQL 就没有安全风险

只读查询仍可能：

• 扫描大量数据导致成本和延迟；
• 泄露敏感字段；
• 通过复杂查询消耗数据库资源；
• 绕过业务过滤条件；
• 把不该暴露的聚合结果返回给用户。

只读解决的是“不能修改”，不解决“能看什么、能扫描多少、结果是否符合业务口径”。

---

十三、生产系统应该把什么交给模型，什么留在宿主侧

一个清晰的权限边界如下：

能力	模型负责	宿主系统负责
理解问题	提取意图、提出假设	记录版本和用户确认
选择字段	根据剖面提出候选	校验字段权限和存在性
生成代码	产生 SQL/Python	语法、策略和安全检查
调用工具	请求调用	授权、参数校验、实际执行
解释结果	组织语言、说明限制	只提供已验证事实
修复错误	根据错误提出修改	限制重试次数和资源
选择图表	提出可视化类型	检查数据契约和可读性
最终发布	撰写报告	验证门禁、审计和脱敏

最重要的边界是：

模型可以提出分析行为，但不能自行决定分析行为已经获得执行权限。

这也是 Function Calling 和 MCP 在工程上的共同含义：模型得到的是能力的描述和调用接口，真正的执行权仍在宿主、工具服务和权限系统中。OpenAI 将工具调用定义为应用侧执行后再把工具输出返回模型的循环；MCP 则通过 Host、Client、Server 和能力协商标准化外部上下文与工具连接。(developers.openai.com)

---

十四、一个可接受的最终报告应该长什么样

对于前面的区域收入问题，最终回答不应只有：

2026 年 1 月华东收入 100 元，华南收入 50 元。

更完整的回答应是：

分析口径：
- 时间范围：[2026-01-01, 2026-02-01)，Asia/Shanghai
- 订单状态：paid
- 收入定义：SUM(amount)
- 订单数定义：COUNT(DISTINCT order_id)

结果：
| 区域 | 收入 | 订单数 |
|---|---:|---:|
| 华东 | 100 | 1 |
| 华南 | 50 | 1 |

验证：
- 区域收入之和 = 过滤后收入总和：通过
- 区域订单数之和 = 有效订单数：通过
- 独立实现重算：通过

复现信息：
- 输入文件：sales.csv
- 输入内容哈希：9d8c...
- 运行 ID：run_7f31
- 分析代码和输出产物：已保存

如果存在无法判断的地方，也要直接暴露：

注意：原始数据未提供退款字段，因此上述收入未扣除退款。

这比模型根据列名猜测“收入已经是净收入”更可靠。

---

结语

数据分析 Agent 的核心能力不是“生成更多代码”，而是建立一条可验证的证据链：

$$\text{文件快照} \rightarrow \text{数据剖面} \rightarrow \text{分析计划} \rightarrow \text{受控代码} \rightarrow \text{结构化产物} \rightarrow \text{独立验证} \rightarrow \text{可复现报告}$$

文件解决“分析什么”；代码执行解决“如何计算”；表格解决“结果是什么”；图表解决“关系如何观察”；复现解决“能否重新得到”；结果验证解决“为什么应该相信”。

缺少其中任何一个环节，系统都可能在技术上完成任务，却无法在工程上证明任务完成。

---

系列导航与关联阅读

• 系列入口：Agent 工程完整路线：从运行循环、记忆与协议到安全、评测和生产交付
• 上一篇：SQL Agent：Schema 上下文、只读约束、查询校验、成本和审计
• 下一篇：客服 Agent：意图、知识、工单、升级、质量和会话记忆
• 延伸：Agent 代码执行沙箱：进程、容器、文件、网络、资源和销毁

官方资料

• OpenAI Function Calling Guide
• Model Context Protocol Specification

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。