modelengine-nexent-agent — 数据处理智能体（ModelEngine 任务一）

一个可运行的数据处理智能体（Data Processing Agent）：接收自然语言的数据处理需求， 自动完成任务理解 → 规划 → 多算子组合调度 → 执行 → 状态跟踪与异常处理，跑通一条典型 ETL / 文本处理流水线。构建在 Nexent 兼容的智能体层 与 DataMate 兼容的算子层 之上， 零重型依赖即可本地跑起来（只需 pandas + PyYAML）。

赛题：ModelEngine 开源贡献赛 · 任务一（基于 Nexent 的数据处理智能体，30 分）。MIT 许可。

---

亮点（对齐评分项）

评分项	分值	本项目对应
架构设计合理性	5	智能体层与数据处理层彻底解耦：agent/（规划/审核/调度）从不直接碰 pandas 或算子内部，只经 DataMate 算子市场 API 与 Nexent 工具契约交互
智能体能力	10	任务理解 + 自主规划：意图关键词 + 实时数据画像联合驱动算子选型/排序，自动推荐清洗/去重/抽取；可选 LLM 规划后端
功能完整性	10	一条完整医疗 ETL 跑通、结果正确；双质检门（计划期端口兼容 + 运行期数据质量硬校验）保证正确性
工程质量	5	文档 + 单测（11 例全绿）+ CLI/Notebook 双 Demo + Docker + 可复现

---

架构

flowchart TB
    NL([自然语言需求]) --> PL[决策组 Planner
任务理解 + 产出类型化 ETL 计划]
    PL --> RV1{质检组 Reviewer 门1
计划期端口兼容校验}
    RV1 -- 封驳 --> PL
    RV1 -- 准入 --> EX[实现组 Executor
按计划调度算子·状态跟踪·异常隔离]
    EX --> RV2{质检组 Reviewer 门2
运行期数据质量硬门}
    RV2 --> OUT[(干净表 / 报告)]
    subgraph 数据处理层 · DataMate 兼容
      OPS[(算子市场 OPERATORS
Mapper / Filter / Slicer)]
    end
    subgraph 智能体层 · Nexent 兼容
      TOOLS[Tool 契约 · OpenAIModel · MessageObserver]
    end
    EX -. OPERATORS.build .-> OPS
    PL & EX -. 事件流 .-> TOOLS

三组分工（现代命名，非三省六部）：

• 决策组 / Planner：把自然语言需求解析成类型化 ETL 计划（每步声明算子、参数、消费/产出列）。计划是结构化产物，不是自由文本。
• 质检组 / Reviewer：两道确定性（非 LLM）门。门 1 在执行前校验算子存在性与相邻步骤列兼容；门 2 在执行后硬校验数据质量（必填列零缺失、主键唯一、目标 schema 齐备、行数合理）。不达标即判失败——由代码判定，不靠模型自评。
• 实现组 / Executor：按计划从 DataMate 算子市场 OPERATORS.build(name, **params) 实例化算子并在样本流上调度，逐步跟踪状态（rows/drop/quarantine/耗时），单条脏数据抛错时隔离该记录而非崩溃。

ETL 流水线（内置医疗样例）

flowchart LR
    SRC[(patients.csv)] --> D1[DropMissing
清洗缺失]
    D1 --> D2[Dedup
去重]
    D2 --> D3[FieldExtract
血压→收缩/舒张]
    D3 --> D4[TypeNormalize
年龄→int·性别归一]
    D4 --> D5[FieldTransform
年龄分组]
    D5 --> D6[TextClean
备注脱敏]
    D6 --> SINK[(clean.csv)]

---

快速开始

# 1) 安装依赖（仅 pandas + PyYAML）
pip install -r requirements.txt

# 2) 一键跑内置医疗 ETL 样例
python -m modelengine_agent.cli --recipe patients

# 3) 跑自定义数据（自然语言目标 + 可选 hints）
python -m modelengine_agent.cli \
  --input data/patients.csv \
  --goal "清洗缺失与重复，规范年龄性别，抽取血压，分组并脱敏" \
  --output output/clean.csv --report output/report.json

# 4) Notebook 演示
jupyter notebook notebooks/demo.ipynb

CLI 会依次打印：数据画像 → 规划链路 → 逐算子执行状态 → 质检门指标 → 结果路径。 退出码 0 表示质检门通过。

真实运行输出（节选）

== 执行状态 (execution) ==
  [0] DropMissing    success  13->10 drop=3 quar=0
  [1] Dedup          success  10->9  drop=1 quar=0
  [2] FieldExtract   success  9->9
  [3] TypeNormalize  partial  9->8  quar=1     # age="forty" 触发异常 → 隔离该记录
  [4] FieldTransform success  8->8
  [5] TextClean      success  8->8
== 质检门 == 通过: True  (missing::age=0, duplicate_keys=0)
== 结果 == 状态: success -> output/clean.csv (8 行)

输出表：age 由 " 52 "/"60岁" 规范为整数；gender 的 female/male/男/女 归一为 F/M； blood_pressure "140/90" 抽取为 systolic/diastolic；派生 age_group；notes 中邮箱/电话脱敏为 <email>/<phone>。

---

智能体如何“自主规划”

规划分两层，诚实区分“推断”与“配置”：

• 结构规划（自动推断）：Planner 先对数据做画像（每列缺失率、数值占比、类别基数、是否 A/B 可拆分、是否性别枚举、文本长度），再结合目标里的意图关键词（清洗/去重/规范/抽取/转换/脱敏），自动决定用哪些算子、按什么顺序、作用在哪些列，并做 schema 投影推导目标列集。
• 领域参数（可选 hints/LLM）：像“血压拆成收缩压/舒张压”“年龄分箱边界”这类领域知识，由可选的 hints 提供，或交给 LLM 后端推断。这样智能体对新数据集会自适应（缺失就补清洗、有重复就补去重、有可拆列就补抽取），而不把一切写死。

LLM 规划后端（可选）：设置环境变量即可让 Nexent OpenAIModel 产出受算子目录约束的 JSON 计划，任何失败自动回退确定性规划（demo 默认离线，不需要它）：

export MODELENGINE_LLM_API_KEY=...  MODELENGINE_LLM_BASE_URL=https://...  MODELENGINE_LLM_MODEL=...
pip install openai
python -m modelengine_agent.cli --recipe patients --llm

---

与 Nexent / DataMate 的集成（诚实交代）

两层均按从官方源码核实的真实接口复刻，接口同形，可无缝接回真库：

数据处理层 = DataMate 算子 ABI（源：ModelEngine-Group/DataMate runtime/python-executor/datamate/core/base_op.py、runtime/ops/）

• 算子继承 Mapper(1:1) / Filter(保留或丢弃) / Slicer(1:N)，实现 execute(sample: Dict) -> ...；
• 全局 OPERATORS = Registry("Operators")，用 DataMate 的 register_module(module_name, module_path) 注册（算子市场）；
• 每个算子是一个目录：process.py + metadata.yml（name/type/description/inputs/outputs/...），与 DataMate 仓库结构一致；
• 其中 TextClean 直接移植 DataMate 真实算子 EmailNumberCleaner 的行为。

智能体层 = Nexent SDK 契约（源：ModelEngine-Group/nexent sdk/nexent/core/...）

• CoreAgent(observer=, tools=, model=, name=, max_steps=) 构造面复刻自 Nexent CoreAgent；
• 工具是 smolagents 风格的 Tool（类属性 name/description/inputs/output_type + forward()），无 @tool 装饰器——与 Nexent 一致；每个 DataMate 算子被包成一个 Nexent 工具（agent/tools.py）；
• OpenAIModel(observer=, model_id=, api_key=, api_base=) 与 Nexent 同形；MessageObserver 复刻其流式观察者角色。

为什么用兼容层而不是直接装真库：真实 nexent SDK 要求 Python >=3.11,<3.12 且依赖 smolagents/mem0，真实 DataMate 运行时含 Java 后端 + Ray + Milvus + PaddleOCR（需 Docker 整栈）。本兼容层无这些约束，任何有 pandas 的环境都能跑，且把待接真库的位置显式标注在各文件顶部——接真库时改基类导入即可（详见 docs/DESIGN.md）。这是工程取舍，不是臆造接口：所有类名/方法名/签名均来自上述源码。

---

项目结构

modelengine_agent/
  datamate_compat/            # 数据处理层（DataMate 算子 ABI）
    base_op.py                #   BaseOp/Mapper/Filter/Slicer + Registry + OPERATORS
    ops/
      filter/{drop_missing,dedup}/{process.py,metadata.yml}
      mapper/{type_normalize,field_extract,field_transform,text_clean}/{process.py,metadata.yml}
  nexent_compat/              # 智能体层（Nexent SDK 契约）
    agent.py  tool.py  model.py  observer.py
  agent/                      # 业务编排
    planner.py                #   决策组：NL + 画像 -> 类型化计划（确定性 + 可选 LLM）
    reviewer.py               #   质检组：计划期 + 运行期两道确定性硬门
    executor.py               #   实现组：算子调度 + 状态跟踪 + 异常隔离
    data_agent.py             #   DataProcessingAgent 编排器
    plan.py                   #   类型化 Plan/PlanStep/QualityGate
  cli.py                      # CLI Demo（含内置 patients 配方）
data/patients.csv             # 真实小样本（含缺失/重复/混杂格式/PII）
notebooks/demo.ipynb          # Notebook Demo
tests/test_pipeline.py        # 11 例单测
Dockerfile                    # 容器化（加分项）

---

算子市场

算子	类型	作用
DropMissing	Filter	必填列缺失/空值/哨兵值即丢弃该行
Dedup	Filter	按主键去重（有状态，保留首现）
FieldExtract	Mapper	正则捕获组把一列拆成多列（血压→收缩/舒张）
TypeNormalize	Mapper	容错数值解析 + 去空白 + 类别归一（性别）
FieldTransform	Mapper	声明式派生（bucket/map/rename/const，无 eval）
TextClean	Mapper	文本列邮箱/电话脱敏（移植 DataMate EmailNumberCleaner）

新增算子 = 放一个目录（process.py+metadata.yml）并在 ops/__init__.py 注册一行。

---

Docker

docker build -t modelengine-nexent-agent .
docker run --rm modelengine-nexent-agent            # 跑内置样例
docker run --rm -v "$PWD/data:/app/data" -v "$PWD/output:/app/output" \
  modelengine-nexent-agent --input data/patients.csv --output output/clean.csv

测试

pip install pytest
python -m pytest -q          # 11 passed

---

路线图（任务二/三，本次不在 MVP 范围）

• 任务二（40 分 + NPU 加分）：医疗知识图谱生成与问答——实体/关系抽取算子复用本仓算子市场契约；NPU 算子优化需昇腾环境，为加分项，后置。
• 任务三（30 分）：数据分析智能体——复用任务一清洗产出做 NL2SQL + BI 可视化。

三任务共享同一套“规划 + 双质检门 + 算子市场”内核，详见 ../方案.md。

许可

MIT。参赛成果按赛题要求以 MIT 开源。