Skip to content

handsomebig2004/AutoNexus

BranchesTags

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

25 Commits
.vscode
.vscode
 
 
config
config
 
 
src
src
 
 
tests
tests
 
 
.gitignore
.gitignore
 
 
README.md
README.md
 
 

Repository files navigation

AutoNexus

autoNexus 是一个面向自动建模流程的多 Agent 项目。当前目标是先让大模型生成多种可运行的建模方案，统一执行后比较结果；模型/Agent 的迭代优化暂时放到后续阶段。

Agent 分工

Agent 职责
requirement_agent 理解用户需求，把问题转化为可建模的问题，并明确所需输入和输出。
research_agent 查找数据与相关文献，生成模型方案，并筛选可用数据和可用方法。
data_agent 根据数据和方法特征完成数据清洗、预处理和特征工程。
train_agent 编写模型训练代码并运行训练。
evaluation_agent 评估实验结果，辅助比较不同方案。
检查的时候考虑使用一个固定的脚本

项目目录结构

.
├── config/                 # YAML 等配置文件
├── src/                    # 项目核心代码
│   ├── agents/             # 各类 Agent 的源代码
│   ├── interfaces/         # CLI、stdin、文件、未来 Web API 等外部输入接口
│   ├── models/             # 总结后的模型信息，供 LLM Agent 参考
│   ├── prompts/            # 固定提示词
│   ├── schemas/            # Agent 输出的标准格式
│   ├── skills/             # 输入给 LLM Agent 的技能文档
│   ├── tools/              # 外部连接、代码执行、数据处理等工具
│   └── utils/              # Agent 和 tools 之间复用的小函数
└── tasks/                  # 每次任务运行产生的任务目录

tasks/ 目录约定

每运行一个 task，可能会产生多个建模方法；每个方法对应一个 run_xxx。

tasks/
└── task_xxx/
    ├── data/
    │   ├── raw/                    # research_agent 找到的原始数据
    │   └── processed/run_xxx/      # data_agent 处理后的数据
    ├── doc/                        # research_agent 找到的文献或资料
    ├── src/
    │   └── run_xxx/                # 大模型为该 run 生成的代码
    │       ├── preprocess/         # data_agent 生成的预处理代码
    │       └── train/              # train_agent 生成的训练代码
    └── outputs/run_xxx/            # 模型运行结果、日志、审查记录等任务内产物

约定：

  • data/raw/ 保存原始数据。
  • data/processed/run_xxx/ 保存某个 run 对应的处理后数据。
  • doc/ 保存文献、数据说明、研究记录等可被其他 Agent 读取的资料。
  • src/run_xxx/ 保存该 run 的生成代码。
  • outputs/run_xxx/ 保存该 run 的输出结果、日志和只在当前 task 内使用的审查记录。

config/ 目录约定

config/ 用来保存各种 YAML 配置文件。后续配置变多后，可以按 Agent、运行环境或任务类型继续分层。

src/ 目录约定

目录 说明
src/agents/ 各类 Agent 的源代码。每个 Agent 应输出固定结构，方便后续工具或其他 Agent 消费。
src/interfaces/ 外部输入接口。负责把命令行、文件、stdin 或未来 Web API 的输入统一转成内部 schema。
src/pipeline/ 流程编排层。负责创建 task 目录、串联 Agent、写状态文件和日志，不负责具体建模逻辑。
src/prompts/ 各类模型的固定提示词。后续可以考虑加入 manager 或 memory，在提示词进入 Agent 前做统一处理。
src/models/ 经过总结的模型信息，可输入给 LLM Agent，辅助它按指定模型写代码。
src/skills/ 输入给 LLM Agent 的技能文档，例如如何找论文、如何使用 models/ 中的模型信息。
src/schemas/ Agent 输出的标准格式定义。
src/tools/ 与外部连接、代码执行、数据处理相关的工具函数或类，后续可加入安全检查。
src/utils/ Agent 和 tools 之间复用的小函数。后续如果做 Agent 记忆，也可以在这里扩展 memory/ 等目录。

输入接口说明

AutoNexus 的输入层和 Agent 层是分离的。命令行、文件、PowerShell 管道、未来网页接口都不应该直接把字符串塞进某个 Agent，而是先统一转换成 UserRequest。

统一输入对象定义在：

src/schemas/user_request.py

结构如下：

{
  "request_text": "预测客户是否流失",
  "source": "stdin",
  "source_path": null,
  "metadata": {
    "interface": "cli"
  }
}

字段含义：

字段 含义
request_text 用户原始需求文本，后续会传给 requirement_agent。
source 输入来源，目前支持 cli、file、stdin、web。
source_path 如果来源是文件，记录文件路径；否则为 null。
metadata 额外信息，例如接口类型、文件编码、用户 ID、前端来源等。

为什么要有输入接口层

这样设计是为了降低耦合：

CLI / 文件 / PowerShell 管道 / Web API
        ↓
UserRequest
        ↓
requirement_agent
        ↓
TaskDefinition

requirement_agent 只接收 UserRequest，不关心用户输入来自命令行、文件还是网页。以后从 CLI 换成 Web API 时，只需要新增 Web 接口，把 HTTP 请求转换成同一个 UserRequest，不用改 Agent 本身。

当前支持的输入方式

当前命令行接口在：

src/interfaces/cli.py

支持三种输入方式。

1. 直接传入文本

conda activate automl
python -m src.interfaces.cli --text "预测客户是否流失"

输出：

{
  "request_text": "预测客户是否流失",
  "source": "cli",
  "source_path": null,
  "metadata": {
    "interface": "cli"
  }
}

适合快速测试或脚本调用。

2. 从文件读取

conda activate automl
python -m src.interfaces.cli --file .\request.txt

默认文件编码是 utf-8-sig，可以兼容带 BOM 的 UTF-8 文件。如果需要指定编码：

python -m src.interfaces.cli --file .\request.txt --encoding utf-8

输出里的 source 会是 file，并记录 source_path：

{
  "request_text": "预测未来7天销量",
  "source": "file",
  "source_path": "request.txt",
  "metadata": {
    "interface": "cli",
    "encoding": "utf-8-sig"
  }
}

适合较长需求、从文档中复制出的任务描述，或者后续批量任务。

3. 从 PowerShell 管道 / stdin 读取

conda activate automl
"给客户做无监督分群" | python -m src.interfaces.cli

也可以：

Get-Content .\request.txt | python -m src.interfaces.cli

输出里的 source 会是 stdin：

{
  "request_text": "给客户做无监督分群",
  "source": "stdin",
  "source_path": null,
  "metadata": {
    "interface": "cli"
  }
}

如果 PowerShell 管道传中文时出现乱码，可以先设置：

[Console]::OutputEncoding = [System.Text.Encoding]::UTF8
$OutputEncoding = [System.Text.Encoding]::UTF8

写出标准输入 JSON

CLI 可以把标准化后的 UserRequest 写入文件：

python -m src.interfaces.cli --text "预测房价" --output tasks\user_request.json

这会同时：

  • 在终端打印 UserRequest
  • 将同样内容写入 tasks/user_request.json

后续 pipeline 可以直接读取这个 JSON，作为 task 的原始输入记录。

如何传给 requirement_agent

后续 requirement_agent 推荐只接收 UserRequest，不要接收裸字符串：

from src.interfaces.cli import read_user_request_from_cli
from src.agents.requirement_agent import RequirementAgent

user_request = read_user_request_from_cli()

agent = RequirementAgent()
task_definition = agent.run(user_request)

requirement_agent 内部应该使用：

user_request.request_text

作为 LLM prompt 的用户需求输入，同时把这些信息写入日志：

user_request.source
user_request.source_path
user_request.metadata

这样可以追踪任务来自哪里，也方便复现实验。

未来 Web API 如何接入

以后如果加网页或 HTTP API，不需要改 requirement_agent，只需要在 Web 层构造同一个 schema：

from src.schemas import UserRequest

user_request = UserRequest(
    request_text=payload["text"],
    source="web",
    metadata={
        "user_id": payload.get("user_id"),
        "session_id": payload.get("session_id"),
    },
)

task_definition = requirement_agent.run(user_request)

也就是说：

CLI 和 Web 的区别只存在于 interfaces 层。
Agent 和 pipeline 只认识 UserRequest。

输入接口的边界

src/interfaces/cli.py 只负责：

  • 读取 --text
  • 读取 --file
  • 读取 stdin / PowerShell 管道
  • 转成 UserRequest
  • 可选写出 JSON

它不负责：

  • 调用 LLM
  • 判断任务类型
  • 创建模型方案
  • 创建 run
  • 执行 pipeline

这些逻辑应该放在 agents/ 或后续的 pipeline/ 中，避免输入接口和业务流程耦合。

requirement_agent 说明

requirement_agent 是 AutoNexus pipeline 的第一个 Agent，也是一个“建模闸门”。它的职责不是直接建模，而是判断用户需求是否可以进入后续自动建模流程。

实现文件：

src/agents/requirement_agent.py
src/prompts/requirement_agent.md
src/schemas/requirement.py
src/utils/requirement_validation.py

输入

requirement_agent 接收标准化后的 UserRequest，而不是裸字符串。

from src.schemas import UserRequest

user_request = UserRequest(
    request_text="我有客户历史消费数据和是否流失标签，想预测客户是否会流失",
    source="cli",
)

字段来源可以是：

  • --text
  • --file
  • PowerShell 管道 / stdin
  • 未来 Web API

只要转换成 UserRequest，后续 requirement_agent 的使用方式都一样。

输出

requirement_agent 输出 TaskDefinition，定义在：

src/schemas/requirement.py

核心结构如下：

{
  "task_name": "客户流失预测",
  "task_type": "classification",
  "decision": "accepted",
  "should_model": true,
  "problem_statement": "基于客户历史消费数据预测客户是否会流失。",
  "input_mode": {
    "data_type": "tabular",
    "required_inputs": ["客户历史消费数据", "客户是否流失标签"],
    "optional_inputs": [],
    "target_column": "churn",
    "id_columns": ["customer_id"],
    "time_column": null,
    "data_granularity": null,
    "known_data_sources": []
  },
  "output_mode": {
    "prediction_type": "class_label",
    "target_description": "客户是否流失",
    "output_format": "每个客户输出一个流失/不流失类别"
  },
  "constraints": {
    "must_have": [],
    "must_not": [],
    "resource_limits": [],
    "privacy_or_safety": []
  },
  "evaluation": {
    "primary_metric": "f1",
    "secondary_metrics": ["accuracy"],
    "validation_strategy": "train/validation/test split",
    "metric_reasoning": "客户流失任务可能存在类别不均衡，f1 比 accuracy 更稳妥。"
  },
  "assumptions": [],
  "missing_information": {
    "critical": [],
    "optional": []
  },
  "user_facing_response": "已识别为分类任务，可以进入后续建模。",
  "downstream_notes": {
    "for_research_agent": [],
    "for_data_agent": [],
    "for_train_agent": [],
    "for_evaluation_agent": []
  },
  "raw_user_request": "我有客户历史消费数据和是否流失标签，想预测客户是否会流失"
}

支持的任务类型

当前只支持四类建模任务：

task_type 含义 示例
classification 预测离散类别、标签、状态、是否发生等。 预测客户是否流失、判断邮件是否垃圾邮件。
regression 预测连续数值。 预测房价、预测评分、预测温度。
forecasting 基于时间顺序预测未来值或趋势。 预测未来 7 天销量、预测下月用电量。
clustering 无监督发现群组或样本结构。 客户分群、用户画像、行为模式发现。

其他需求统一归为：

task_type = unknown
decision = rejected
should_model = false

例如：

  • 只要求写报告
  • 只要求查资料
  • 只要求做可视化
  • 只要求搭系统
  • 没有明确机器学习建模目标

这些不会进入后续建模。

决策状态

requirement_agent 不只是判断 task_type，还会输出 decision。

decision 是否进入建模 含义
accepted 是 任务类型支持，且关键信息和可选确认都已满足。
need_info 否 任务类型支持，但缺少关键建模信息，必须让用户补充。
need_confirmation 暂停 关键信息足够，但缺少可选信息；用户可以补充，也可以回复 yes 直接继续。
rejected 否 不属于当前支持的四类建模任务。

后续 pipeline 推荐这样判断：

task_definition = requirement_agent.run(user_request)

if task_definition.decision == "accepted":
    continue_pipeline(task_definition)

elif task_definition.decision == "need_confirmation":
    return task_definition.user_facing_response

else:
    return task_definition.user_facing_response

critical / optional 缺失信息

missing_information 被拆成两类：

{
  "critical": [],
  "optional": []
}

含义：

字段 是否阻塞建模 含义
critical 是 不补就不能安全建模的信息。
optional 需要用户确认 补了会更好，但用户确认后可以继续建模。

例如 forecasting 任务：

{
  "decision": "need_info",
  "should_model": false,
  "missing_information": {
    "critical": [
      "请说明要预测的目标变量。",
      "请说明时间字段或时间粒度。",
      "请说明希望预测未来多久。"
    ],
    "optional": [
      "如果有节假日、促销、价格等外生变量，可以一起提供。"
    ]
  }
}

这种情况必须补充 critical，不能进入建模。

如果只有 optional 缺失：

{
  "decision": "need_confirmation",
  "should_model": false,
  "missing_information": {
    "critical": [],
    "optional": [
      "建议补充类别分布，用于判断是否需要处理类别不均衡。"
    ]
  }
}

这种情况会先返回用户。用户可以补充信息，也可以回复：

yes

然后系统可以直接进入后续建模，不需要再次调用 requirement_agent。

用户 yes 后如何继续

如果 decision == "need_confirmation"，并且用户回复 yes，使用：

from src.utils import confirm_optional_information

task_definition = confirm_optional_information(task_definition)

这个函数会：

  • 检查当前任务确实是 need_confirmation
  • 确认 critical 为空
  • 将 decision 改成 accepted
  • 将 should_model 改成 true
  • 清空 missing_information
  • 把“用户确认缺少可选信息仍继续建模”写入 assumptions

它不会重新调用 LLM，也不会重新跑 requirement_agent。

内部处理逻辑

RequirementAgent.run() 的流程：

UserRequest
  ↓
读取 src/prompts/requirement_agent.md
  ↓
拼接用户需求
  ↓
调用 LLMClient.generate()
  ↓
记录 llm_calls.jsonl
  ↓
从 LLM 输出中提取 JSON
  ↓
使用 TaskDefinition 做 schema 校验
  ↓
使用 normalize_requirement_gate() 做确定性业务校验
  ↓
必要时把 unsafe accepted 降级为 need_info / need_confirmation
  ↓
写 task_definition.json
  ↓
写 task.jsonl 摘要
  ↓
返回 TaskDefinition

其中：

  • LLM 负责理解自然语言。
  • TaskDefinition 负责结构校验。
  • normalize_requirement_gate() 负责最终闸门校验。
  • 只有 accepted 才允许进入后续 Agent。

确定性验证逻辑

确定性验证在：

src/utils/requirement_validation.py

主要函数：

validate_requirement_for_modeling(task_definition)
normalize_requirement_gate(task_definition)
confirm_optional_information(task_definition)

验证内容包括：

  • task_type 必须是四类支持任务之一。
  • decision 必须是 accepted 才能直接建模。
  • should_model 必须为 true 才能直接建模。
  • missing_information.critical 必须为空。
  • missing_information.optional 必须为空，或者经过用户 yes 确认。
  • input_mode.data_type 不能是 unknown。
  • input_mode.required_inputs 不能为空。
  • classification / regression 必须有 target_column。
  • forecasting 必须有 target_column 和 time_column。
  • output_mode.prediction_type 不能是 unknown。
  • output_mode.target_description 不能为空。
  • output_mode.output_format 不能为空。
  • evaluation.primary_metric 不能为空。

如果 LLM 输出 accepted，但确定性验证发现关键信息不全，会自动变成：

decision = need_info
should_model = false

如果只有可选信息缺失，会自动变成：

decision = need_confirmation
should_model = false

如何使用 requirement_agent

最小用法：

from src.agents.requirement_agent import RequirementAgent
from src.schemas import UserRequest

user_request = UserRequest(
    request_text="我有客户历史消费数据和是否流失标签，想预测客户是否会流失",
    source="cli",
)

agent = RequirementAgent()
task_definition = agent.run(user_request)
print(task_definition.decision)

带 task 日志和输出文件：

from src.agents.requirement_agent import RequirementAgent
from src.schemas import UserRequest
from src.utils import TaskLogger

task_dir = "tasks/task_0"
logger = TaskLogger(task_dir)

agent = RequirementAgent(task_logger=logger)
task_definition = agent.run(
    UserRequest(
        request_text="我有客户历史消费数据和是否流失标签，想预测客户是否会流失",
        source="cli",
    ),
    output_path="tasks/task_0/metadata/task_definition.json",
)

如果要从 CLI 输入接入：

from src.interfaces.cli import read_user_request_from_cli
from src.agents.requirement_agent import RequirementAgent

user_request = read_user_request_from_cli()
task_definition = RequirementAgent().run(user_request)

requirement_agent 相关文件

文件 作用
src/agents/requirement_agent.py Agent 主体逻辑：调用 LLM、解析输出、校验、写日志和输出。
src/prompts/requirement_agent.md 固定提示词：说明支持任务、决策规则、输出 JSON 格式。
src/schemas/requirement.py TaskDefinition schema：约束输出结构和基础规则。
src/schemas/user_request.py UserRequest schema：统一 CLI、文件、stdin、Web 输入。
src/utils/requirement_validation.py 确定性业务校验：判断是否能进入后续建模。
src/utils/text.py 从 LLM 回复中提取 JSON。
src/utils/ids.py 生成 llm_0001 这类 LLM 调用编号。
src/tools/llm_client.py 访问外部 LLM。
src/utils/task_logger.py 写入 task.jsonl 和 llm_calls.jsonl。

pipeline runner 说明

当前已经实现了一个最小版 pipeline runner，文件在：

src/pipeline/runner.py

它现在只负责跑第一段链路：

UserRequest
  -> 创建 tasks/task_xxx
  -> 写 user_request.json
  -> 创建 TaskLogger
  -> 调用 requirement_agent
  -> 写 task_definition.json
  -> 写 task_state.json
  -> 根据 decision 决定当前状态

也就是说，目前 runner 不会直接调用 research_agent、data_agent、train_agent 或 evaluation_agent。这是刻意保守的设计：先把任务创建、需求理解、状态落盘和日志跑通，再逐步往后接更多 Agent。

runner 输入

runner 的输入仍然是标准化后的 UserRequest：

from src.pipeline import PipelineRunner
from src.schemas import UserRequest

runner = PipelineRunner()
result = runner.run_requirement_stage(
    UserRequest(
        request_text="我有客户历史消费数据和是否流失标签，想预测客户是否会流失",
        source="cli",
    )
)

run_requirement_stage() 和 run() 当前等价，都会执行最小 pipeline。

runner 输出

runner 返回 PipelineResult：

print(result.task_id)
print(result.task_dir)
print(result.status)
print(result.can_continue)
print(result.user_facing_response)

字段含义：

字段 含义
task_id 本次任务编号，例如 task_0。
task_dir 本次任务目录，例如 tasks/task_0。
status 当前任务状态。
current_agent 当前停在哪个 Agent，目前是 requirement_agent。
task_definition requirement_agent 输出的 TaskDefinition。
task_state_path task_state.json 路径。
task_definition_path task_definition.json 路径。
user_request_path user_request.json 路径。
can_continue 是否可以进入后续建模 Agent。
user_facing_response 可以返回给用户看的说明。

runner 创建的文件

每次运行会创建一个新的 task 目录：

tasks/
└── task_xxx/
    ├── user_request.json
    ├── task_definition.json
    ├── task_state.json
    └── logs/
        ├── task.jsonl
        └── llm_calls.jsonl

这些文件的职责：

文件 作用
user_request.json 保存标准化后的原始用户输入，方便复现任务。
task_definition.json 保存 requirement_agent 结构化后的建模任务定义。
task_state.json 保存当前任务状态快照，方便 CLI、Web API 或前端快速读取。
logs/task.jsonl 保存 task 级摘要日志。
logs/llm_calls.jsonl 保存完整 LLM 调用日志。

task_state.json

task_state.json 是当前状态快照，不替代 JSONL 日志。日志适合追溯全过程，state 文件适合快速判断任务现在停在哪里。

示例：

{
  "task_id": "task_0",
  "status": "ready_for_modeling",
  "current_agent": "requirement_agent",
  "created_at": "2026-05-08T12:00:00+08:00",
  "updated_at": "2026-05-08T12:00:00+08:00",
  "duration_seconds": 2.35,
  "input": {
    "source": "cli",
    "source_path": null
  },
  "requirement": {
    "task_name": "客户流失预测",
    "task_type": "classification",
    "decision": "accepted",
    "should_model": true,
    "critical_missing_information_count": 0,
    "optional_missing_information_count": 0
  },
  "next_step": "Continue with research_agent or data_agent.",
  "paths": {
    "user_request": "user_request.json",
    "task_definition": "task_definition.json",
    "task_log": "logs/task.jsonl",
    "llm_calls_log": "logs/llm_calls.jsonl"
  }
}

当前状态值

runner 会根据 TaskDefinition.decision 生成任务状态：

status 来源 含义
ready_for_modeling decision=accepted 需求闸门通过，后续可以接 research_agent 或 data_agent。
waiting_for_required_information decision=need_info 缺少 critical 信息，必须让用户补充。
waiting_for_optional_confirmation decision=need_confirmation 只缺 optional 信息，用户可补充，也可回复 yes 继续。
rejected decision=rejected 不属于当前支持的建模任务。
failed runner 或 Agent 异常 pipeline 执行失败，需要查看日志。

如何和 CLI 输入配合

当前可以先在 Python 里把 CLI 输入转成 UserRequest，再传给 runner：

from src.interfaces.cli import read_user_request_from_cli
from src.pipeline import PipelineRunner

user_request = read_user_request_from_cli()
result = PipelineRunner().run(user_request)

print(result.status)
print(result.user_facing_response)

PowerShell 示例：

conda activate automl
"我有客户历史消费数据和是否流失标签，想预测客户是否会流失" | python your_script.py

后续可以再补一个正式的 scripts/run_pipeline.py 或 src/interfaces/pipeline_cli.py，把这段调用封装成命令行入口。

后续扩展方式

后面继续扩展时，runner 应该只做编排，不把具体 Agent 逻辑写进去。

推荐扩展方向：

if result.can_continue:
    -> research_agent
    -> data_agent
    -> 为每个候选方案创建 run_xxx
    -> train_agent
    -> evaluation_agent

每接入一个新 Agent，建议同时补三类内容：

  • 对应 schema，约束 Agent 输出。
  • 对应 prompt，约束 LLM 行为。
  • 对应自动化测试，用 fake LLM 或 fake agent，不直接访问真实 API。

自动化测试说明

当前项目使用 pytest 做自动化测试。测试目录在：

tests/
├── agents/
├── pipeline/
└── utils/

运行测试

先进入 conda 环境：

conda activate automl

运行全部测试：

python -m pytest

只运行 utils 测试：

python -m pytest tests\utils

只运行 requirement_agent 测试：

python -m pytest tests\agents\test_requirement_agent.py

只运行 data_agent 测试：

python -m pytest tests\agents\test_data_agent.py

只运行 DataProcessPlan schema 测试：

python -m pytest tests\schemas\test_data_plan.py

只运行 pipeline runner 测试：

python -m pytest tests\pipeline

当前已验证通过的测试范围：

python -m pytest tests\schemas tests\tools tests\agents tests\pipeline tests\utils

当前测试覆盖范围

测试目录 覆盖内容
tests/schemas/test_data_plan.py DataProcessPlan 的状态、字段必填、切分策略、缺失值策略、额外字段拒绝和 JSON 序列化。
tests/utils/ IO、配置读取、文本解析、ID 生成、编号目录、日志事件、TaskLogger、RunLogger、requirement gate。
tests/agents/test_requirement_agent.py requirement_agent 的解析、校验、降级、日志和错误处理。
tests/agents/test_data_agent.py data_agent 的计划解析、need_info、输出文件、task/run 日志和错误处理。
tests/tools/ 数据读取、数据画像、字段推断和数据质量检查等确定性工具。
tests/pipeline/test_runner.py pipeline runner 的 task 创建、状态写入、异常状态记录。

为什么测试里不调用真实 LLM

自动化测试里不直接访问 OpenAI 或 DeepSeek。原因是：

  • 真实 API 会产生费用。
  • 网络和限流会让测试不稳定。
  • LLM 输出有随机性，不适合做稳定断言。
  • 单元测试应该验证代码逻辑，不应该依赖外部服务。

因此，requirement_agent 测试使用 FakeLLMClient，pipeline runner 测试使用 fake requirement agent。这样可以稳定验证：

  • LLM 返回合法 JSON 时是否能解析。
  • LLM 返回坏 JSON 时是否能报错。
  • 缺少关键信息时是否会降级为 need_info。
  • 缺少 optional 信息时是否会变成 need_confirmation。
  • 日志和状态文件是否写到正确位置。

手动测试和自动化测试的边界

自动化测试不访问真实 LLM，但仍然需要手动测试真实 provider 配置是否可用。

建议后续单独新增：

scripts/manual_test_llm_client.py
scripts/manual_test_requirement_agent.py

这类脚本可以真实调用 API，用来检查：

  • config/settings.yaml 是否配置正确。
  • OPENAI_API_KEY 或 DEEPSEEK_API_KEY 是否可用。
  • prompt 在真实模型上的输出质量。
  • LLM 返回结果是否符合 schema。

这些脚本不应该放进自动化测试，也不应该作为 CI 的默认检查项。

新增测试的原则

后续每写一个 Agent 或 tool，建议至少补这些测试：

  • 正常输入能得到预期结构。
  • 缺少必要字段时会失败或降级。
  • 输出文件写到正确路径。
  • 日志写入正确事件。
  • 外部依赖使用 fake/mock，不直接访问网络。

对于会生成代码或运行脚本的模块，还要额外测试：

  • 生成代码路径是否正确。
  • 运行失败时是否记录 stderr 和错误类型。
  • 不会覆盖其他 run 或 task 的文件。

data_profiler 说明

data_profiler 是给 data_agent 使用的数据概览工具，文件在：

src/tools/data_profiler.py

它的职责是客观描述一个表格数据集“长什么样”，不负责决定怎么清洗，也不负责生成预处理代码。后续推荐流程是：

data_loader
  -> data_profiler
  -> schema_infer
  -> data_quality
  -> data_agent

也就是说，在让 LLM 写预处理脚本之前，先用 data_profiler 生成稳定、可复现、JSON 可序列化的数据报告。

输入

profile_table() 支持三类输入：

输入类型 说明
pandas.DataFrame 直接分析已经在内存里的 DataFrame。
LoadedTable 复用 data_loader.load_table() 已经加载好的结果，避免重复读文件。
str / Path 传入数据文件路径，内部会复用 load_table() 读取。

最小用法：

from src.tools.data_profiler import profile_table

profile = profile_table("tasks/task_0/data/raw/train.csv")

如果已经有 DataFrame：

profile = profile_table(dataframe, dataset_name="train")

如果已经使用 data_loader：

from src.tools.data_loader import load_table
from src.tools.data_profiler import profile_table

loaded = load_table("tasks/task_0/data/raw/train.csv")
profile = profile_table(loaded)

参数

参数 作用
dataset_name 数据集显示名称；如果输入是路径，默认使用文件名。
max_sample_values 每列最多保存多少个非空样例值，默认 5。
max_top_values 每列最多保存多少个高频值，默认 10。
max_columns 最多详细分析多少列；用于超宽表，避免 prompt 太长。
include_value_counts 是否输出每列高频值统计，默认 True。
**load_kwargs 当输入是路径时，透传给 load_table()，例如 encoding。

输出

输出是普通 dict，可以直接：

  • 写入 JSON 文件。
  • 传给 schema_infer.py。
  • 传给 data_quality.py。
  • 放进 data_agent prompt。

输出结构示例：

{
  "dataset_name": "train.csv",
  "source": {
    "source_type": "path",
    "path": "tasks/task_0/data/raw/train.csv",
    "file_name": "train.csv",
    "file_extension": ".csv",
    "shape": {
      "rows": 10000,
      "columns": 8
    }
  },
  "shape": {
    "rows": 10000,
    "columns": 8,
    "profiled_columns": 8,
    "truncated_columns": []
  },
  "columns": [
    {
      "name": "age",
      "dtype": "float64",
      "pandas_type": "float",
      "non_null_count": 9800,
      "missing_count": 200,
      "missing_rate": 0.02,
      "unique_count": 61,
      "unique_rate": 0.006224,
      "sample_values": [23.0, 45.0, 31.0],
      "top_values": [
        {
          "value": 32.0,
          "count": 310,
          "rate": 0.031633
        }
      ],
      "numeric_summary": {
        "min": 18.0,
        "max": 80.0,
        "mean": 41.2,
        "median": 39.0,
        "std": 12.5
      }
    }
  ],
  "duplicate_row_count": 12,
  "memory_usage_bytes": 1536000
}

顶层字段含义：

字段 含义
dataset_name 数据集名称。
source 输入来源信息；路径输入会包含文件名、扩展名、文件大小等。
shape 数据集行列数、实际分析列数、被截断的列名。
columns 每一列的详细画像。
duplicate_row_count 完全重复行数量。
memory_usage_bytes DataFrame 内存占用。

列级字段含义：

字段 含义
name 列名。
dtype pandas 原始 dtype。
pandas_type 归一化后的 pandas 类型，如 integer、float、string、datetime、boolean、object。
non_null_count 非空数量。
missing_count 缺失数量。
missing_rate 缺失比例。
unique_count 非空唯一值数量。
unique_rate 唯一值数量 / 非空数量。
sample_values 去重后的非空样例值。
top_values 高频值、数量和比例。
numeric_summary 数值列的最小值、最大值、均值、中位数、标准差。
datetime_summary 时间列或可解析为时间的字符串列的最小时间和最大时间。

设计边界

data_profiler 只做确定性统计，不做这些事：

  • 不判断哪一列一定是 target。
  • 不决定哪些列必须删除。
  • 不生成预处理脚本。
  • 不调用 LLM。
  • 不修改原始数据。

这些判断后续分别交给：

模块 职责
schema_infer.py 根据 profile 推断字段角色，例如 ID、target、feature、time。
data_quality.py 根据 profile 和 schema 检查数据质量风险。
data_agent 根据任务、profile、schema、quality report 生成预处理方案或脚本。

自动化测试

对应测试文件：

tests/tools/test_data_profiler.py

当前测试覆盖：

  • 能分析 DataFrame 的行列数、缺失率、唯一值和数值摘要。
  • 能从文件路径读取数据并保留来源信息。
  • 能接收 LoadedTable，避免重复读文件。
  • 能用 max_columns 限制详细分析列数。
  • 输出可以直接 JSON 序列化。

运行方式：

conda activate automl
python -m pytest tests\tools\test_data_profiler.py

schema_infer 说明

schema_infer 是字段语义和建模角色推断工具，文件在：

src/tools/schema_infer.py

它接收 data_profiler 的输出，推断每一列大概应该怎么用，例如：

  • 哪些列像 ID，应该默认丢弃。
  • 哪些列是数值特征。
  • 哪些列是类别特征。
  • 哪些列是自由文本。
  • 哪些列是时间字段。
  • 哪一列是 target。
  • 哪些列只是 target 候选，需要用户或后续 Agent 确认。

它同样不调用 LLM，也不修改数据，只做确定性启发式判断。

输入

最主要输入是 profile_table() 的输出：

from src.tools.data_profiler import profile_table
from src.tools.schema_infer import infer_schema

profile = profile_table("tasks/task_0/data/raw/train.csv")
schema = infer_schema(profile)

如果已经有 TaskDefinition，推荐传入：

schema = infer_schema(profile, task_definition=task_definition)

这样 TaskDefinition.input_mode.target_column 和 TaskDefinition.input_mode.time_column 会优先于启发式规则。

也可以手动指定：

schema = infer_schema(
    profile,
    target_column="churn",
    time_column="date",
)

参数

参数 作用
profile data_profiler.profile_table() 生成的数据画像。
task_definition 可选的 TaskDefinition 或 dict，用于读取 task_type、target_column、time_column。
target_column 可选的显式目标列，会覆盖 task_definition 中的目标列。
time_column 可选的显式时间列，会覆盖 task_definition 中的时间列。
high_cardinality_threshold 判断高唯一率字符串列是否像 ID 的阈值，默认 0.8。
categorical_unique_threshold 判断低基数类别列的唯一值数量阈值，默认 20。

输出

输出是普通 dict，可以直接写 JSON 或传给 data_agent。

示例：

{
  "dataset_name": "train.csv",
  "task_type": "classification",
  "columns": [
    {
      "name": "customer_id",
      "semantic_type": "id",
      "role": "identifier",
      "recommended_use": "drop",
      "confidence": 0.9,
      "reasons": [
        "column name or high unique rate indicates identifier"
      ]
    },
    {
      "name": "age",
      "semantic_type": "numeric",
      "role": "feature",
      "recommended_use": "use",
      "confidence": 0.8,
      "reasons": [
        "inferred from pandas_type=integer"
      ]
    },
    {
      "name": "churn",
      "semantic_type": "boolean",
      "role": "target",
      "recommended_use": "target",
      "confidence": 1.0,
      "reasons": [
        "matches explicit target_column"
      ]
    }
  ],
  "feature_columns": ["age"],
  "target_column": "churn",
  "target_candidates": [],
  "id_columns": ["customer_id"],
  "time_columns": [],
  "drop_columns": ["customer_id"],
  "warnings": []
}

顶层字段含义：

字段 含义
dataset_name 数据集名称，来自 profile。
task_type 建模任务类型，如果传入了 TaskDefinition 就会记录。
columns 每一列的推断结果。
feature_columns 推荐作为特征输入的列。
target_column 明确目标列；如果只是猜测，不会直接填这里。
target_candidates 目标列候选，需要用户或后续 Agent 确认。
id_columns 识别出的 ID 或 identifier 列。
time_columns 识别出的时间索引列。
drop_columns 建议默认丢弃的列。
warnings 推断时发现的不确定或冲突情况。

列级字段含义：

字段 含义
name 列名。
semantic_type 字段语义类型。
role 建模角色。
recommended_use 推荐使用方式。
confidence 启发式置信度。
reasons 推断理由，方便调试和给 LLM 解释上下文。

semantic_type

当前支持的语义类型：

semantic_type 含义
id ID、主键、编号类字段。
numeric 普通数值字段。
boolean 二值字段，例如 0/1、true/false。
categorical 低基数类别字段。
high_cardinality_categorical 高基数字符串类别字段，需要谨慎编码。
text 自由文本字段。
datetime 时间字段。
unknown 暂时无法判断。

role 和 recommended_use

role 表示建模角色：

role 含义
feature 可作为特征。
target 明确目标列。
target_candidate 可能是目标列，但没有显式确认。
identifier ID 字段。
time_index 时间索引字段。

recommended_use 表示推荐动作：

recommended_use 含义
use 作为普通特征使用。
use_text 作为文本特征使用，后续需要文本处理。
target 作为目标列。
time_index 作为时间索引。
drop 默认丢弃，例如 ID。
review 不确定，需要后续确认。

推断规则边界

schema_infer 会优先相信显式信息：

target_column / time_column 参数
  > task_definition.input_mode.target_column / time_column
  > 列名和 profile 启发式判断

它不会把“像 target 的列”直接当成 target，除非用户或 TaskDefinition 明确指定。比如列名叫 label，但没有显式 target_column，它会进入：

{
  "target_candidates": ["label"]
}

这样可以避免自动化 pipeline 在目标列不明确时误建模。

自动化测试

对应测试文件：

tests/tools/test_schema_infer.py

当前测试覆盖：

  • 显式 target_column 优先。
  • ID 列会被识别为 identifier/drop。
  • forecasting 任务的 time_column 会被识别为 time_index。
  • 没有显式 target 时，只生成 target_candidates。
  • 文本列和高基数字符串列能被区分。
  • clustering 任务不会强制要求 target。
  • 输出可以直接 JSON 序列化。

运行方式：

conda activate automl
python -m pytest tests\tools\test_schema_infer.py

data_quality 说明

data_quality 是数据质量检查工具，文件在：

src/tools/data_quality.py

它接收 data_profiler 的输出，以及可选的 schema_infer 输出和 TaskDefinition，检查数据是否存在会影响自动建模的风险。

推荐调用顺序：

data_loader
  -> data_profiler
  -> schema_infer
  -> data_quality
  -> data_agent

它不调用 LLM，也不修改数据，只输出结构化质量报告。后续 pipeline 可以根据 can_continue 和 blocking_issue_count 决定是否继续。

输入

最小用法：

from src.tools.data_profiler import profile_table
from src.tools.data_quality import check_data_quality

profile = profile_table("tasks/task_0/data/raw/train.csv")
quality = check_data_quality(profile)

推荐用法：

from src.tools.data_profiler import profile_table
from src.tools.schema_infer import infer_schema
from src.tools.data_quality import check_data_quality

profile = profile_table("tasks/task_0/data/raw/train.csv")
schema = infer_schema(profile, task_definition=task_definition)
quality = check_data_quality(
    profile,
    schema=schema,
    task_definition=task_definition,
)

参数

参数 作用
profile data_profiler.profile_table() 生成的数据画像。
schema 可选的 schema_infer.infer_schema() 输出，用于知道 target、feature、ID、time 等角色。
task_definition 可选的 TaskDefinition 或 dict，用于知道任务类型和显式 target/time。
high_missing_rate 高缺失率阈值，默认 0.4。
class_imbalance_rate 分类任务类别极度不均衡阈值，默认 0.9。
min_rows_warning 样本量过少 warning 阈值，默认 50。

输出

输出是普通 dict，可以直接写 JSON 或传给 data_agent。

示例：

{
  "dataset_name": "train.csv",
  "task_type": "classification",
  "issues": [
    {
      "issue_type": "target_missing_values",
      "severity": "error",
      "columns": ["churn"],
      "message": "Target column churn contains missing values.",
      "recommendation": "Drop rows with missing target or ask the user to provide complete labels.",
      "blocking": true,
      "evidence": {
        "missing_rate": 0.05
      }
    },
    {
      "issue_type": "class_imbalance",
      "severity": "warning",
      "columns": ["churn"],
      "message": "Target column churn is highly imbalanced.",
      "recommendation": "Use stratified splitting and imbalance-aware metrics or resampling.",
      "blocking": false,
      "evidence": {
        "top_value": 0,
        "top_rate": 0.94,
        "unique_count": 2
      }
    }
  ],
  "summary": {
    "error_count": 1,
    "warning_count": 1,
    "info_count": 0,
    "blocking_issue_count": 1
  },
  "recommended_actions": [
    "Drop rows with missing target or ask the user to provide complete labels."
  ],
  "can_continue": false
}

顶层字段含义：

字段 含义
dataset_name 数据集名称，来自 profile。
task_type 建模任务类型，来自 schema 或 TaskDefinition。
issues 具体数据质量问题列表。
summary 错误、警告、阻塞问题数量统计。
recommended_actions 优先建议动作；如果有阻塞问题，只返回阻塞问题建议。
can_continue 是否可以继续进入后续建模；只要有 blocking issue 就是 false。

issue 字段

字段 含义
issue_type 机器可读的问题类型。
severity 严重程度，当前使用 error 和 warning。
columns 涉及的列名；数据集级问题可以为空。
message 给人看的简短说明。
recommendation 建议如何处理。
blocking 是否阻止后续自动建模。
evidence 结构化证据，例如缺失率、重复率、类别占比。

当前检查项

issue_type 严重程度 是否阻塞 含义
empty_dataset error 是 数据集没有任何行。
too_few_rows warning 否 样本量低于 min_rows_warning。
duplicate_rows warning 否 存在完全重复行。
all_missing_column error 是 某列全为空。
high_missing_rate warning 否 某列缺失率高于阈值。
constant_column warning 否 某列只有一个非空唯一值。
no_feature_columns error 是 schema 中没有可用特征列。
target_missing error 是 监督任务没有 target。
target_not_found error 是 指定 target 不在 profile 中。
target_missing_values error 是 target 列存在缺失值。
target_single_class error 是 分类 target 少于两个类别。
class_imbalance warning 否 分类 target 极度不均衡。
time_index_missing error 是 forecasting 任务没有时间索引列。
time_index_not_parseable warning 否 时间列没有被 profiler 识别为可解析时间。
high_cardinality_categorical warning 否 高基数类别特征，需要谨慎编码。
possible_id_leakage warning 否 ID 字段被放入特征，可能造成泄漏或过拟合。

设计边界

data_quality 只负责发现风险，不负责修复风险。

它不会：

  • 删除重复行。
  • 填补缺失值。
  • 删除常量列。
  • 重采样类别不均衡数据。
  • 生成预处理脚本。

这些动作应该由后续 data_agent 根据 quality report 生成预处理方案或代码。

如何给 data_agent 使用

建议传给 data_agent 的上下文至少包含：

{
  "task_definition": "...",
  "data_profile": "...",
  "inferred_schema": "...",
  "quality_report": "..."
}

data_agent 应该重点读取：

  • quality_report.can_continue
  • quality_report.issues
  • quality_report.recommended_actions
  • schema.feature_columns
  • schema.target_column
  • schema.drop_columns
  • profile.columns

如果 can_continue=false，后续 pipeline 可以选择：

  • 直接打回用户补充或修正数据。
  • 让 data_agent 只生成修复建议，不生成训练脚本。
  • 在用户确认后继续，但要把 blocking issue 写入日志和 assumptions。

自动化测试

对应测试文件：

tests/tools/test_data_quality.py

当前测试覆盖：

  • target 缺失值会生成阻塞错误。
  • 分类 target 单类别会阻塞。
  • 分类 target 极度不均衡会 warning。
  • 高缺失列、重复行、常量列会被记录。
  • 监督任务没有 target 会阻塞。
  • forecasting 任务缺少 time_index 会阻塞。
  • 高基数类别特征会 warning。
  • 输出可以直接 JSON 序列化。

运行方式：

conda activate automl
python -m pytest tests\tools\test_data_quality.py

DataProcessPlan 说明

DataProcessPlan 是 data_agent 的结构化输出 schema，文件在：

src/schemas/data_plan.py

它的作用是让 data_agent 先输出“预处理计划”，而不是一上来直接写 preprocess.py。这样可以先审查计划是否合理，再进入后续代码生成阶段。

当前设计里，DataProcessPlan 描述这些内容：

  • 输入数据文件有哪些。
  • 哪一列是 target。
  • 哪些列是 feature。
  • 哪些列是 ID，需要丢弃。
  • 哪些列需要缺失值处理。
  • 哪些类别列需要编码。
  • 哪些数值列需要缩放。
  • 哪些文本列、时间列需要特殊处理。
  • 如何切分 train / validation / test。
  • 预处理阶段应该产出哪些文件。
  • 哪些 quality issue 需要处理。
  • 哪些信息需要打回用户补充。

状态

DataProcessPlan.status 有三种：

status 含义
executable 信息足够，后续可以根据这个计划生成预处理代码。
need_info 缺少关键信息或有阻塞性质量问题，需要用户补充或确认。
rejected 当前数据无法支持该建模任务。

executable 不代表已经完成预处理，只代表“计划可以进入代码生成阶段”。

核心字段

字段 含义
plan_name 计划名称。
status 计划状态。
task_type 建模任务类型。
input_files 输入数据文件路径。
target_column 目标列；分类、回归、预测任务必须有。
time_column 时间列；forecasting 任务必须有。
id_columns ID 或 identifier 列。
feature_columns 可用于建模的特征列。
drop_columns 预处理时建议丢弃的列。
missing_value_plan 缺失值处理计划。
categorical_encoding_plan 类别编码计划。
numeric_scaling_plan 数值缩放计划。
text_processing_plan 文本列处理计划。
datetime_processing_plan 时间列处理计划。
split_strategy 数据切分策略。
column_actions 额外列操作，例如 drop、rename、cast。
output_artifacts 预处理阶段预期产物。
quality_issues_to_handle 需要处理的数据质量问题。
assumptions 做计划时使用的假设。
warnings 不阻塞但需要注意的问题。
user_facing_response 返回给用户看的说明。
downstream_notes 给后续 Agent 的说明。

示例

{
  "plan_name": "Customer churn preprocessing",
  "status": "executable",
  "task_type": "classification",
  "input_files": ["tasks/task_0/data/raw/train.csv"],
  "target_column": "churn",
  "time_column": null,
  "id_columns": ["customer_id"],
  "feature_columns": ["age", "gender", "tenure"],
  "drop_columns": ["customer_id"],
  "missing_value_plan": [
    {
      "columns": ["age"],
      "strategy": "median",
      "fill_value": null,
      "reason": "Age has a small missing rate."
    }
  ],
  "categorical_encoding_plan": [
    {
      "columns": ["gender"],
      "encoding": "one_hot",
      "handle_unknown": "ignore",
      "reason": "Gender is low cardinality.",
      "params": {}
    }
  ],
  "numeric_scaling_plan": [
    {
      "columns": ["age", "tenure"],
      "scaling": "standard",
      "reason": "Use a general scaling default.",
      "params": {}
    }
  ],
  "text_processing_plan": [],
  "datetime_processing_plan": [],
  "split_strategy": {
    "strategy": "stratified",
    "train_size": 0.7,
    "validation_size": 0.1,
    "test_size": 0.2,
    "random_state": 42,
    "stratify_column": "churn",
    "time_column": null,
    "reason": "Classification task should preserve label distribution."
  },
  "column_actions": [
    {
      "columns": ["customer_id"],
      "action": "drop",
      "reason": "Identifier column should not be used as a feature.",
      "params": {}
    }
  ],
  "output_artifacts": [
    {
      "artifact_type": "train_data",
      "path": "data/processed/train.csv",
      "description": "processed training data",
      "required": true
    },
    {
      "artifact_type": "feature_report",
      "path": "metadata/feature_report.json",
      "description": "feature metadata for train_agent",
      "required": true
    }
  ],
  "quality_issues_to_handle": ["class_imbalance"],
  "assumptions": [],
  "warnings": [],
  "user_facing_response": "Data preprocessing plan is ready.",
  "downstream_notes": {
    "for_train_agent": ["Use metadata/feature_report.json."]
  }
}

校验规则

当前 schema 会做这些确定性校验：

规则 说明
executable 必须有 input_files 没有输入数据不能生成可执行计划。
分类、回归、预测任务必须有 target_column 防止没有标签就进入训练。
forecasting 必须有 time_column 时间序列任务不能缺少时间索引。
executable 必须有 feature_columns 没有特征不能训练。
executable 必须有 output_artifacts 后续 Agent 需要知道预处理产物在哪里。
constant 缺失值策略必须有 fill_value 防止填充值不明确。
stratified 切分必须有 stratify_column 分层切分必须知道按哪列分层。
time_based 切分必须有 time_column 时间切分必须知道时间列。
need_info / rejected 必须有 user_facing_response 需要能向用户解释为什么不能继续。
禁止额外字段 避免 LLM 随意扩展 schema。

自动化测试

对应测试文件：

tests/schemas/test_data_plan.py

运行方式：

conda activate automl
python -m pytest tests\schemas\test_data_plan.py

data_agent 说明

data_agent 是数据处理阶段的 LLM Agent，文件在：

src/agents/data_agent.py
src/prompts/data_agent.md

当前版本只输出 DataProcessPlan，不写代码、不运行脚本。这是为了把 data_agent 拆成两个阶段：

阶段 1：理解数据上下文，输出 DataProcessPlan
阶段 2：根据 DataProcessPlan 生成 preprocess.py 并运行

现在只实现了阶段 1。

输入

DataAgent.run() 接收四类上下文：

输入 来源 作用
task_definition requirement_agent 建模任务定义，包括任务类型、target、time、评价指标等。
data_profile data_profiler 数据集基础画像，包括列类型、缺失率、样例值等。
inferred_schema schema_infer 字段角色推断，包括 feature、target、ID、time 等。
quality_report data_quality 数据质量报告，包括阻塞问题和建议动作。

调用示例：

from src.agents.data_agent import DataAgent

agent = DataAgent()
plan = agent.run(
    task_definition=task_definition,
    data_profile=data_profile,
    inferred_schema=inferred_schema,
    quality_report=quality_report,
    output_path="tasks/task_0/runs/run_0/metadata/data_plan.json",
)

带日志：

from src.agents.data_agent import DataAgent
from src.utils import RunLogger, TaskLogger

task_logger = TaskLogger("tasks/task_0")
run_logger = RunLogger("tasks/task_0/runs/run_0")

agent = DataAgent(
    task_logger=task_logger,
    run_logger=run_logger,
)

plan = agent.run(
    task_definition=task_definition,
    data_profile=data_profile,
    inferred_schema=inferred_schema,
    quality_report=quality_report,
    output_path="tasks/task_0/runs/run_0/metadata/data_plan.json",
)

输出

输出是通过 schema 校验后的：

DataProcessPlan

如果传入 output_path，会额外写出：

data_plan.json

内部处理逻辑

DataAgent.run() 的流程：

task_definition + data_profile + inferred_schema + quality_report
  -> 读取 src/prompts/data_agent.md
  -> 拼成 Context JSON
  -> 调用 LLMClient.generate()
  -> 写 llm_calls.jsonl
  -> 从 LLM 输出中提取 JSON
  -> 用 DataProcessPlan 做 schema 校验
  -> 可选写 data_plan.json
  -> 写 task.jsonl / run.jsonl 摘要
  -> 返回 DataProcessPlan

如果 LLM 返回非 JSON，或 JSON 不符合 DataProcessPlan，会抛出：

LLMOutputParseError

提示词原则

src/prompts/data_agent.md 约束了这些行为：

  • 只输出 JSON，不输出 Markdown。
  • 当前阶段不写 Python 代码。
  • 不生成 shell 命令。
  • 不编造不存在的文件、列或标签。
  • 优先相信 data_profiler、schema_infer、data_quality 的确定性报告。
  • 如果 quality_report.can_continue=false，优先返回 need_info。
  • 如果监督任务没有确认 target，返回 need_info。
  • 如果 forecasting 没有确认 time column，返回 need_info。
  • 不要把 target 或 ID 放进 feature_columns。

日志

如果传入 TaskLogger，会写：

tasks/task_xxx/logs/task.jsonl
tasks/task_xxx/logs/llm_calls.jsonl

如果传入 RunLogger，会写：

tasks/task_xxx/runs/run_xxx/logs/run.jsonl

llm_calls.jsonl 会保存完整 prompt 和 response。task.jsonl / run.jsonl 只保存摘要，例如：

  • agent_started
  • llm_call_finished
  • agent_finished
  • agent_failed

自动化测试

对应测试文件：

tests/agents/test_data_agent.py

当前测试不调用真实 LLM，而是使用 FakeLLMClient。

覆盖内容：

  • 合法 LLM 输出可以生成 DataProcessPlan。
  • 可以写出 data_plan.json。
  • blocking quality report 时可以返回 need_info。
  • 可以写 task 级日志、run 级日志和完整 LLM 调用日志。
  • LLM 返回非 JSON 时会抛出 LLMOutputParseError。

运行方式：

conda activate automl
python -m pytest tests\agents\test_data_agent.py

工具规划

只考虑了 data_agent，其他的后面再说（划掉的是目前已实现的）

Tool 文件 作用
LLM访问工具 src/tools/llm_client.py 访问外部LLM
数据读取工具 src/tools/data_loader.py 统一读取 csv、xlsx、json、parquet，返回 DataFrame 和基础信息。
数据概览工具 src/tools/data_profiler.py 统计行列数、字段类型、缺失率、唯一值、样例值、高频值、数值摘要、时间范围、重复行等；也可由 data_agent 调用，用来总结数据集概况。
Schema 推断工具 src/tools/schema_infer.py 根据 data_profiler 输出推断字段语义和建模角色，如数值、类别、文本、时间、ID、target、target 候选等。
数据质量检查工具 src/tools/data_quality.py 根据 profile 和 schema 检查重复行、常量列、高缺失列、target 问题、类别不均衡、时间索引缺失、高基数类别和泄漏风险等。
预处理代码生成辅助 src/tools/code_writer.py 将 LLM 输出的 preprocess.py 安全写入 run 目录。
预处理代码校验工具 src/tools/code_validator.py 检查生成脚本是否包含规定入口函数、危险导入、危险系统调用等；可以后置实现。
预处理代码运行工具 src/tools/code_runner.py 在指定 run 目录运行 preprocess.py，捕获 stdout、stderr 和退出码。
产物管理工具 src/tools/artifact_manager.py 管理 processed data、feature report、preprocessor、日志路径等产物。
数据切分工具 src/tools/splitter.py 提供标准 train/val/test split，并支持分类任务 stratify。
特征报告工具 src/tools/feature_report.py 生成 feature_report.json，供 train_agent 使用。

Utils 规划

Util 文件 作用
路径工具 src/utils/numbered_paths.py 创建 path/prefix_xxx 目录：扫描已有编号，取最大编号并创建下一个目录；使用 filelock 防止并发重复，返回已创建好的 Path。
配置读取 src/utils/config.py 读取 settings.yaml及其中的Agent 配置和默认参数。
JSON/YAML IO src/utils/io.py 统一读写 json、yaml、txt、md，避免各处重复实现。
任务日志工具 src/utils/task_logger.py 写入 task.jsonl 和 llm_calls.jsonl。
run 日志工具 src/utils/run_logger.py 写入某个 run 的 run.jsonl。
文本格式工具 src/utils/text.py 清理 LLM 输出，例如从回复中提取 Python 代码块或 JSON。
ID 生成工具 src/utils/ids.py 生成 llm_0001 等稳定编号，也可从 JSONL 中扫描下一个编号。
哈希/版本工具 src/utils/hash.py 记录原始数据 hash 和脚本 hash，方便复现。
错误类型定义 src/utils/errors.py 定义 DataValidationError、GeneratedCodeError 等异常类型。
日志事件/内容生成 src/utils/log_events.py 生成日志事件里需要的结构化信息。
需求闸门验证 src/utils/requirement_validation.py 验证 TaskDefinition 是否能进入建模，并处理 optional 信息确认。

错误处理规划

错误类型集中定义在 src/utils/errors.py，其他模块只负责在合适的边界捕获并转化错误。

模块 处理方式
src/utils/errors.py 只定义错误类型。
src/tools/llm_client.py 捕获 API 错误，并转成 LLMError。
schema_parser.py / agent.py 捕获输出解析失败，并转成 LLMOutputParseError。
src/tools/code_validator.py 检查生成代码，失败时抛出 GeneratedCodeValidationError。
src/tools/code_runner.py 脚本运行失败时抛出 GeneratedCodeExecutionError。
pipeline/runner.py 根据错误类型决定重试、终止、降级或继续执行。

日志说明

日志系统采用 JSONL 格式，即一行一个 JSON 对象。这样既可以直接打开查看，也方便后续用程序统计、筛选和生成实验报告。

当前日志分为三类：

tasks/
└── task_xxx/
    ├── logs/
    │   ├── task.jsonl        # task 级摘要日志
    │   └── llm_calls.jsonl   # LLM 完整调用日志
    └── runs/
        └── run_xxx/
            └── logs/
                └── run.jsonl # run 级执行日志

三类日志的职责不同：

日志文件 粒度 主要用途
task.jsonl 整个任务 记录任务级时间线、Agent 开始/结束、run 创建/完成、错误摘要、最优方案等。
llm_calls.jsonl LLM 调用 记录完整 prompt、system prompt、response、模型、provider、token 使用量和错误信息。
run.jsonl 单个 run 记录某一个模型方案的代码生成、代码校验、预处理、训练、评估、指标和产物。

通用日志格式

task.jsonl 和 run.jsonl 都使用统一事件结构：

{
  "time": "2026-04-28T12:00:00+08:00",
  "level": "INFO",
  "event": "agent_finished",
  "task_id": "task_0",
  "run_id": "run_0",
  "agent": "data_agent",
  "message": "Data agent finished.",
  "data": {
    "duration_seconds": 3.42,
    "output_path": "tasks/task_0/runs/run_0/metadata/feature_report.json"
  }
}

字段含义：

字段 含义
time 带时区的 ISO 时间戳。
level 日志级别：DEBUG、INFO、WARNING、ERROR、CRITICAL。
event 机器可读的事件名，例如 agent_finished、train_failed。
task_id 当前任务编号，例如 task_0。
run_id 当前 run 编号；task 级事件如果不属于某个 run，可以为 null。
agent 产生事件的 Agent，例如 data_agent、train_agent。
message 给人看的简短说明。
data 结构化附加信息，例如路径、指标、耗时、错误类型等。

task 级摘要日志：task.jsonl

路径：

tasks/task_xxx/logs/task.jsonl

task.jsonl 只记录摘要，不保存完整 prompt 或完整模型回复。它用于快速了解一个 task 从创建到结束发生了什么。

建议记录内容：

类型 事件名示例 记录内容
task 生命周期 task_created、task_started、task_finished、task_failed task 创建、开始、结束、失败原因。
Agent 生命周期 agent_started、agent_finished、agent_failed 哪个 Agent 开始/结束、耗时、输出文件路径、错误摘要。
LLM 调用摘要 llm_call_finished、llm_call_failed llm_call_id、provider、model、status、token 使用量。
run 管理 run_created、run_started、run_finished、run_failed 创建了哪些 run、每个 run 是否成功、失败位置。
结果汇总 best_run_selected、task_report_written 最优 run、关键指标、最终报告路径。

示例：

{"time":"2026-04-28T12:00:00+08:00","level":"INFO","event":"run_finished","task_id":"task_0","run_id":"run_1","agent":"evaluation_agent","message":"Run finished.","data":{"accuracy":0.91,"metrics_path":"tasks/task_0/runs/run_1/metadata/metrics.json"}}

用途：

  • 快速查看 task 整体进展。
  • 判断失败发生在哪个 Agent 或哪个 run。
  • 汇总多个 run 的最终状态。
  • 为后续自动生成 task 总结报告提供结构化依据。

对应工具：

from src.utils import TaskLogger

logger = TaskLogger("tasks/task_0")
logger.info("task_created", message="Task created.")
logger.info("run_created", run_id="run_0", data={"model": "random_forest"})

LLM 完整调用日志：llm_calls.jsonl

路径：

tasks/task_xxx/logs/llm_calls.jsonl

llm_calls.jsonl 用来保存完整 LLM 调用记录。它和 task.jsonl 分开，是因为 prompt 和 response 可能很长；如果全部塞进 task.jsonl，会让 task 摘要变得难读。

建议记录内容：

字段 含义
llm_call_id 本次 LLM 调用的唯一编号，例如 llm_0001。
task_id 所属 task。
run_id 如果这次调用属于某个 run，就记录 run 编号。
agent 调用 LLM 的 Agent。
provider 例如 openai、deepseek。
model 实际使用的模型名。
status success 或 failed。
system_prompt 完整 system prompt。
prompt 完整 user prompt。
response 完整模型输出。
error 失败时的错误类型和错误消息。
usage token 使用量、耗时等可选信息。
metadata 其他辅助信息，例如 prompt 模板版本、schema 名称。

示例：

{"time":"2026-04-28T12:00:00+08:00","llm_call_id":"llm_0001","task_id":"task_0","run_id":"run_0","agent":"data_agent","provider":"deepseek","model":"deepseek-v4-flash","status":"success","system_prompt":"...","prompt":"...","response":"...","error":null,"usage":{"input_tokens":1200,"output_tokens":800},"metadata":{"schema":"PreprocessPlan"}}

用途：

  • 复盘某个 Agent 当时为什么做出某个决策。
  • 调试 LLM 输出格式错误、代码生成错误、幻觉问题。
  • 比较不同 prompt 或模型版本的输出质量。
  • 为后续 prompt 优化和 Agent 迭代提供数据。

对应工具：

from src.utils import TaskLogger

logger = TaskLogger("tasks/task_0")
logger.log_llm_call(
    llm_call_id="llm_0001",
    agent="data_agent",
    provider="deepseek",
    model="deepseek-v4-flash",
    prompt="...",
    response="...",
    run_id="run_0",
)

调用 log_llm_call() 时，默认会同时向 task.jsonl 写入一条简短摘要 llm_call_finished，完整内容仍然只保存在 llm_calls.jsonl。

run 级执行日志：run.jsonl

路径：

tasks/task_xxx/runs/run_xxx/logs/run.jsonl

run.jsonl 记录某一个模型方案的执行细节。它不保存完整 LLM prompt/response，只记录这个 run 的代码、数据、训练、评估和产物状态。

建议记录内容：

类型 事件名示例 记录内容
run 生命周期 run_created、run_started、run_finished、run_failed run 状态、耗时、失败原因。
方案信息 model_plan_loaded 模型名称、模型类型、超参数、方案来源。
代码生成 generated_code_written preprocess.py、train.py、evaluate.py 路径和 hash。
代码校验 generated_code_validated、generated_code_validation_failed 入口函数、危险导入、危险系统调用检查结果。
预处理 preprocess_started、preprocess_finished、preprocess_failed 输入数据路径、输出数据路径、样本数量、特征数量、耗时。
训练 train_started、train_finished、train_failed 模型文件路径、训练耗时、最佳参数、验证集指标。
评估 evaluate_started、evaluate_finished、evaluate_failed 测试集指标、评估报告路径、图表路径。
产物 artifact_written 模型、数据、图、报告、metrics 文件路径。
指标 metric_recorded accuracy、f1、rmse、auc 等指标。
错误 script_failed、run_failed 错误类型、错误消息、是否可重试。

示例：

{"time":"2026-04-28T12:00:00+08:00","level":"INFO","event":"train_finished","task_id":"task_0","run_id":"run_0","agent":"train_agent","message":"Training finished.","data":{"model_path":"tasks/task_0/runs/run_0/artifacts/models/model.pkl","duration_seconds":42.18,"metrics":{"accuracy":0.91,"f1":0.89}}}

用途：

  • 精确复盘某个模型方案是如何执行的。
  • 定位失败发生在预处理、训练、评估还是代码校验阶段。
  • 记录每个 run 的模型文件、数据文件、报告文件和指标文件。
  • 后续自动比较多个 run，生成实验表格或最优方案报告。

对应工具：

from src.utils import RunLogger

logger = RunLogger("tasks/task_0/runs/run_0")
logger.info("run_started", agent="train_agent")
logger.generated_code_written("preprocess", "generated/preprocess.py", agent="data_agent")
logger.artifact_written("model", "artifacts/models/model.pkl", agent="train_agent")
logger.metric_recorded({"accuracy": 0.91, "f1": 0.89}, split="test")

三类日志如何配合

三类日志通过 task_id、run_id 和 llm_call_id 关联：

task.jsonl
  看到 run_0 的 data_agent 调用了一次 LLM：llm_call_id=llm_0001

llm_calls.jsonl
  用 llm_call_id=llm_0001 找到完整 prompt 和 response

run.jsonl
  用 run_id=run_0 查看该模型方案后续代码生成、预处理、训练、评估过程

推荐排查顺序：

  1. 先看 task.jsonl，判断整个任务在哪一步失败。
  2. 如果失败和某个 run 有关，再看对应 runs/run_xxx/logs/run.jsonl。
  3. 如果失败和 LLM 输出有关，再用 llm_call_id 查 llm_calls.jsonl。
  4. 如果失败和脚本运行有关，后续可再查看 preprocess.stdout.log、train.stderr.log 等脚本级日志。

当前已实现的日志相关工具

工具 文件 作用
build_log_event() src/utils/log_events.py 生成统一结构的日志事件。
build_error_event() src/utils/log_events.py 根据异常生成统一结构的错误事件。
EventTimer src/utils/log_events.py 记录 Agent、run 或脚本执行耗时。
append_jsonl() src/utils/io.py 向 JSONL 文件追加一条 JSON 记录。
TaskLogger src/utils/task_logger.py 写入 task.jsonl 和 llm_calls.jsonl。
RunLogger src/utils/run_logger.py 写入某个 run 的 run.jsonl。

About

AutoNexus: A Multi-Agent Pipeline for Machine and Deep Synergistic Learning，or AutoNexus-MDSL

Resources

Readme
Activity

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

No releases published

Packages

 
 
 

Contributors

Languages