Data Interpreter：数据分析与代码执行 Agent

学习目标

理解 Data Interpreter 如何实现自主编写、执行和迭代代码，以及 Plan-and-Execute 架构在数据分析场景的应用。

---

项目实践

Data Interpreter 概览

Data Interpreter（简称 DI）是 MetaGPT 中最灵活的 Agent，能够：

• 根据自然语言需求自主编写 Python 代码
• 在 Jupyter Notebook 中执行代码并获取结果
• 根据执行结果迭代修改代码
• 使用外部工具（搜索引擎、API 等）增强能力

from metagpt.roles.di.data_interpreter import DataInterpreter

di = DataInterpreter()
await di.run("Run data analysis on sklearn Iris dataset, include a plot")

关键配置

class DataInterpreter(Role):
    name: str = "David"
    profile: str = "DataInterpreter"
    auto_run: bool = True           # 自动运行，无需人工确认
    use_plan: bool = True           # 使用计划模式
    use_reflection: bool = False    # 失败后使用反思
    execute_code: ExecuteNbCode     # Jupyter 代码执行器
    tools: list[str] = []           # 可用工具列表
    react_mode: Literal["plan_and_act", "react"] = "plan_and_act"
    max_react_loop: int = 10        # react 模式最大循环次数

Plan-and-Act 架构

_plan_and_act() 是 DI 的核心执行逻辑：

async def _plan_and_act(self) -> Message:
    self._set_state(0)
    try:
        rsp = await super()._plan_and_act()  # 调用 Role 的 plan_and_act
        await self.execute_code.terminate()   # 清理 kernel
        return rsp
    except Exception as e:
        await self.execute_code.terminate()
        raise e

Role._plan_and_act() 的流程（在 role.py 中定义）：

async def _plan_and_act(self) -> Message:
    # 1. 创建计划
    goal = self.rc.memory.get()[-1].content
    await self.planner.update_plan(goal=goal)

    # 2. 顺序执行任务
    while self.planner.current_task:
        task = self.planner.current_task
        task_result = await self._act_on_task(task)
        await self.planner.process_task_result(task_result)

    # 3. 返回结果
    return self.planner.get_useful_memories()[0]

DI 通过 _act_on_task() 实现具体执行：

async def _act_on_task(self, current_task: Task) -> TaskResult:
    code, result, is_success = await self._write_and_exec_code()
    return TaskResult(code=code, result=result, is_success=is_success)

编写-执行循环

_write_and_exec_code() 是 DI 最关键的循环：

async def _write_and_exec_code(self, max_retry: int = 3):
    counter = 0
    success = False

    while not success and counter < max_retry:
        ### 1. 编写代码 ###
        code, cause_by = await self._write_code(counter, plan_status, tool_info)
        self.working_memory.add(Message(content=code, role="assistant"))

        ### 2. 执行代码 ###
        result, success = await self.execute_code.run(code)
        print(result)
        self.working_memory.add(Message(content=result, role="user"))

        ### 3. 处理结果 ###
        counter += 1

每个步骤的输入：

• plan_status：当前计划状态（已完成任务、代码、结果）
• tool_info：推荐工具的 schema 信息（BM25 推荐）
• working_memory：代码和执行的对话历史

ExecuteNbCode：Jupyter 执行器

ExecuteNbCode 通过 nbclient 连接 Jupyter kernel 执行代码：

class ExecuteNbCode(Action):
    async def run(self, code: str) -> tuple[str, bool]:
        # 1. 将代码发送到 Jupyter kernel
        # 2. 获取执行结果（stdout/stderr/图表）
        # 3. 返回结果和成功/失败标志
        return result, success

关键特性：

• Kernel 是持久化的，变量状态跨次执行保持
• 支持 Python 代码和 bash 命令
• 图表输出以 base64 编码返回

TaskType：任务类型系统

DI 使用 TaskType 枚举对任务分类：

class TaskType(Enum):
    DATA_PREPROCESS = "data_preprocess"     # 数据预处理
    FEATURE_ENGINEERING = "feature_engineering"  # 特征工程
    MODEL_TRAIN = "model_train"             # 模型训练
    # ...

不同任务类型影响：

• Planner.get_plan_status() 生成的 prompt（包含 guidance 字段）
• CheckData 动作在数据预处理/特征工程/模型训练阶段检查数据变化

React 模式备选

除了 plan_and_act，DI 也支持标准 react 模式：

async def _think(self) -> bool:
    """react 模式：LLM 判断是否需要继续行动"""
    prompt = REACT_THINK_PROMPT.format(
        user_requirement=self.user_requirement,
        context=context
    )
    rsp = await self.llm.aask(prompt)
    rsp_dict = json.loads(CodeParser.parse_code(text=rsp))
    # {"thoughts": "...", "state": true/false}
    need_action = rsp_dict["state"]
    self._set_state(0) if need_action else self._set_state(-1)
    return need_action

---

问题与规避

Jupyter Kernel 残留状态

• 持久化 kernel 意味着变量跨次执行保持
• 如果代码定义了一个变量但后续执行中未清除，可能被后续代码误用
• 对策：在代码开始时显式重置变量，或使用 execute_code.terminate() 后重建 kernel

max_retry 限制

• 默认 max_retry=3，3 次失败后不再尝试
• 对于复杂任务可能需要更多重试
• 对策：增大 max_retry 或启用 use_reflection=True 让 LLM 反思失败原因

Plan 过度拆分

• Planner 可能将简单任务拆分为过多子任务
• 每个子任务都有 LLM 调用，增加成本
• 对策：简单任务直接使用 react 模式而非 plan_and_act

---

设计取舍

Plan-and-Act vs React

维度	Plan-and-Act	React
适用场景	多步骤复杂任务	简单/探索性任务
成本	高（额外 Plan 生成）	低
可控性	高（计划可审查）	低（LLM 自由发挥）
调试性	好（任务边界清晰）	差（连续循环）

Jupyter vs 沙箱执行

• Jupyter 适合交互式数据分析，状态持久化是优点
• 沙箱（Docker）适合代码生成，状态隔离是优点
• DI 选择 Jupyter 因为其主要场景是数据分析

---

参考来源

• 源码验证: metagpt/roles/di/data_interpreter.py:36 — DataInterpreter 类
• 源码验证: metagpt/roles/di/data_interpreter.py:107 — _write_and_exec_code() 循环
• 源码验证: metagpt/actions/di/execute_nb_code.py — ExecuteNbCode 执行器
• 源码验证: metagpt/strategy/planner.py:58 — Planner 类
• 源码验证: metagpt/strategy/task_type.py — TaskType 枚举