# DataPilot

DataPilot 是一个使用 Python 构建的智能数据分析 Agent 原型。用户上传 CSV 或 Excel
文件后，可以查看数据结构与质量报告、执行只读 DuckDB SQL、用自然语言进行
Text-to-SQL 查询，以及让 LangGraph 把复杂问题拆成多个步骤并生成分析报告和一张 PNG
图表。

当前项目重点展示可验证的分层设计、MCP 工具复用、受控 SQL、LangGraph 工作流和
SQLite Checkpoint。它是一个本地单机原型，不是面向生产环境的完整数据平台。

## 已完成能力

- 读取 `.csv` 和 `.xlsx`，Excel 默认使用第一个工作表；
- 确定性识别 Excel 标题行、多行表头、合并单元格、空行空列和明确汇总行；
- 使用 Pandas 识别字段类型、缺失值、重复行及基础统计；
- 将 DataFrame 注册到进程内 DuckDB，并返回唯一 `dataset_id`；
- 只允许一条 `SELECT` 或 `WITH ... SELECT`，结果最多返回 1000 行；
- 提供 FastAPI 数据集、查询和 Agent API；
- 提供无需 React 的本地浏览器界面，自动完成上传、提问和结果展示；
- 通过 MCP Server 暴露 Schema、Profile、Preview 和只读 SQL 工具；
- 使用 Pydantic Structured Outputs 生成 DuckDB SQL；
- 使用 LangGraph 执行单步骤 Text-to-SQL，并最多修复 SQL 两次；
- 使用父 Planning Graph 将问题拆成 1～4 个独立步骤；
- 从第一个适合绘图的成功步骤生成一张 bar 或 line PNG；
- 使用 `AsyncSqliteSaver` 持久化父图 Checkpoint；
- 通过 `thread_id` 查询最新状态和最多 50 条历史摘要；
- 使用 Mock LLM 完成无网络的全量 pytest 和端到端验收。

## 技术栈

- Python 3.11+
- FastAPI、Pydantic
- Pandas、DuckDB
- MCP Python SDK
- LangChain、LangGraph
- DeepSeek Chat Completions API
- Matplotlib
- AsyncSqliteSaver、SQLite
- pytest

## 技术架构

```mermaid
flowchart TD
    Client["HTTP / MCP 客户端"] --> API["FastAPI API 层"]
    API --> DatasetService["DatasetService"]
    API --> QueryService["QueryService"]
    API --> AgentServices["Graph Services"]

    DatasetService --> Loader["DatasetLoader"]
    Loader --> ExcelDetector["ExcelStructureDetector"]
    DatasetService --> Profiler["DatasetProfiler"]
    DatasetService --> DuckDB["DuckDBService"]
    QueryService --> DuckDB

    AgentServices --> PlanningGraph["Planning Graph"]
    PlanningGraph --> TextToSQLGraph["Text-to-SQL 子图"]
    TextToSQLGraph --> MCPClient["DataPilotMCPClient"]
    MCPClient --> MCPServer["MCP Server"]
    MCPServer --> DatasetService
    MCPServer --> QueryService

    PlanningGraph --> ChartService["ChartService"]
    PlanningGraph --> SQLite["AsyncSqliteSaver"]
```

各层职责：

- API 层：HTTP 参数校验、依赖获取和业务异常到状态码的转换；
- Service 层：组织上传、Profile、查询和 Graph 调用；
- Loader/Profiler/DuckDB/Chart：确定性底层能力，不调用 LLM；
- ExcelStructureDetector：识别工作表结构并生成分析用标准化副本，不修改原文件；
- MCP 层：把已有 Service 暴露成工具，不复制业务逻辑；
- Text-to-SQL 子图：Schema、SQL 生成、查询、修复和回答；
- Planning 父图：规划步骤、逐步调用子图、保存结果、生成图表和报告；
- Checkpointer：按 `thread_id` 保存父图各 super-step 的状态快照。

## 项目结构

```text
app/
├── api/                 # FastAPI 路由
├── graph/               # 单步骤 Text-to-SQL 子图
├── llm/                 # LLM 协议和 DeepSeek 客户端
├── models/              # Pydantic 请求、响应和领域模型
├── planning/            # 多步骤 Planning 父图
├── services/            # 数据、SQL、图表和 Graph Service
├── web/                 # 本地 HTML、CSS、JavaScript 交互页面
├── main.py              # 应用组合根与 lifespan
├── mcp_client.py        # MCP 工具调用封装
└── mcp_server.py        # MCP 工具定义
tests/                   # 单元、API、工作流和项目验收测试
sample_data/             # 可复现演示数据
```

## Excel 结构识别

上传 `.xlsx` 后，`DatasetLoader` 仍然只读取第一个工作表，但不再直接把第一行固定当作
表头。它先调用 `ExcelStructureDetector`，按照确定性规则完成：

- 跳过表头前只有标题内容的行；
- 将最多 3 行的合并式表头组合成唯一字段名；
- 只展开数据区真实存在的纵向合并单元格，例如分组或类别标签；
- 删除全空数据行、全空数据列和标签明确为“合计、总计、小计、汇总、平均”的汇总行；
- 清理字段和文本两侧空格，为重复或空白字段生成稳定名称；
- 返回工作表名、标题、表头位置、清理计数、置信度和 warnings。

原始 Excel 文件不会被修改。CSV 继续沿用原有读取方式。结构结果位于上传响应的
`summary.excel_structure`；置信度为 `low` 时应先核对 Preview。当前规则不会猜测字段的
业务含义，也不会在多个工作表或同一工作表的多个独立表格之间自动选择。

`app/graph/` 和 `app/planning/` 不是重复目录：前者处理一个自然语言查询，后者负责把
复杂请求拆成多个子查询。`TextToSQLService` 是保留并经过测试的线性基线；当前 HTTP
Agent 接口使用的是 `TextToSQLGraphService`。

## LangGraph 工作流

### 单步骤 Text-to-SQL 子图

```mermaid
flowchart LR
    Start["START"] --> Schema["load_schema"]
    Schema -->|成功| Generate["generate_sql"]
    Schema -->|失败| Failed["mark_failed"]
    Generate --> Execute["execute_sql"]
    Execute -->|成功| Answer["generate_answer"]
    Execute -->|失败且未超过 2 次| Repair["repair_sql"]
    Repair --> Execute
    Execute -->|修复耗尽| Failed
    Answer --> End["END"]
    Failed --> End
```

Schema 获取和 SQL 执行必须经过 MCP。LLM 只负责生成结构化 SQL、修复 SQL 和组织自然
语言回答，最终 SQL 仍由 DuckDB 查询层执行安全校验。

### 多步骤 Planning 父图

```mermaid
flowchart LR
    Start["START"] --> Schema["load_schema"]
    Schema --> Plan["create_plan"]
    Plan --> Select["select_next_step"]
    Select --> Child["execute_current_step"]
    Child --> Store["store_step_result"]
    Store -->|仍有步骤| Select
    Store -->|全部完成| Chart["generate_chart"]
    Chart --> Report["synthesize_report"]
    Report --> End["END"]
```

Planner 最多生成 4 个互相独立的自然语言步骤。单个步骤失败不会阻止后续步骤；至少一个
步骤成功时，整体仍可为 `completed`。图表失败只写入 `chart_error`，不会改变步骤状态。

### Checkpoint

父图使用 `AsyncSqliteSaver`，默认数据库为：

```text
.datapilot/checkpoints/checkpoints.sqlite
```

历史固定按最新到最旧返回，最多 50 条。服务重启后可以读取旧 State 和 History，但
不会自动 Resume，也不会重新注册旧 DuckDB 数据集。

## MCP 工具

MCP Streamable HTTP 服务挂载在：

```text
http://127.0.0.1:8000/mcp
```

| 工具 | 输入 | 输出 | 职责 |
|---|---|---|---|
| `get_dataset_schema` | `dataset_id` | 字段和 DuckDB 类型 | 查询 Schema |
| `get_dataset_profile` | `dataset_id` | 数据质量 Profile | 查询确定性质量报告 |
| `preview_dataset` | `dataset_id`, `limit` | 最多 5 行 | 预览数据 |
| `execute_readonly_sql` | `dataset_id`, `sql` | 最多 1000 行 | 执行受控只读 SQL |

MCP 工具只调用 `DatasetService` 和 `QueryService`，不重新实现文件读取、Profile 或 SQL。

## 安装与配置

```bash
python -m pip install -e ".[dev]"
```

环境变量示例见 `.env.example`：

```env
DEEPSEEK_API_KEY=
DEEPSEEK_BASE_URL=https://api.deepseek.com
DEEPSEEK_MODEL=deepseek-v4-flash
CHECKPOINT_DB_PATH=.datapilot/checkpoints/checkpoints.sqlite
LANGGRAPH_STRICT_MSGPACK=true
```

- `DEEPSEEK_API_KEY`：DeepSeek 控制台生成的密钥；未配置时 Agent API 返回 503；
- `DEEPSEEK_BASE_URL`：默认使用 DeepSeek 官方兼容地址；
- `DEEPSEEK_MODEL`：默认使用 `deepseek-v4-flash`，可改为账户可用的模型；
- `CHECKPOINT_DB_PATH`：服务端配置，客户端不能修改；相对路径按启动目录解析；
- `LANGGRAPH_STRICT_MSGPACK`：限制 checkpoint 反序列化类型；项目未启用 pickle fallback。

项目不会自动读取 `.env` 文件。在 VSCode 的 CMD 终端中，应分两行执行：

```cmd
set "DEEPSEEK_API_KEY=你的新密钥"
python -m uvicorn app.main:app --reload
```

PowerShell 使用：

```powershell
$env:DEEPSEEK_API_KEY="你的新密钥"
python -m uvicorn app.main:app --reload
```

密钥必须在启动 Uvicorn 前设置。重启终端后需要重新设置，且不要把真实密钥写入
README、`.env.example`、截图或 Git。

`DeepSeekLLMClient` 使用官方 OpenAI 兼容的 Chat Completions 接口。普通回答读取
`message.content`；SQL 和 AnalysisPlan 等结构化结果使用 JSON Output，再由 Pydantic
执行严格模型校验。项目不再使用 OpenAI Responses API。

启动服务：

```bash
uvicorn app.main:app --reload
```

OpenAPI 文档：`http://127.0.0.1:8000/docs`

本地操作页面：`http://127.0.0.1:8000/`

## 本地操作页面

启动 DataPilot 后，直接在浏览器打开：

```text
http://127.0.0.1:8000/
```

页面支持：

- 点击选择或拖入 CSV/XLSX 文件；
- 自动保存上传响应中的 `dataset_id`，无需手动复制；
- 显示文件名称、行列数、缺失值、重复行和质量警告；
- 展示结构识别后的前 5 行数据；
- 使用“直接查询”调用单步骤 Text-to-SQL；
- 使用“综合分析”调用多步骤 Planning Graph；
- 展示自然语言回答、SQL、查询结果、分析步骤和生成的 PNG 图表；
- 在同一浏览器标签页刷新时恢复当前数据集。

浏览器只保存 `dataset_id`，数据仍由当前 FastAPI 进程管理。服务重启后旧
`dataset_id` 会失效，需要重新上传文件。页面不加载外部前端资源，也不改变现有 API、
MCP 或 Agent 的调用边界。

## API

| 方法 | 路径 | 作用 |
|---|---|---|
| GET | `/health` | 健康检查 |
| GET | `/` | 本地上传和自然语言分析页面 |
| POST | `/api/datasets/upload` | 上传 CSV/XLSX |
| GET | `/api/datasets/{dataset_id}` | 概况和 Profile |
| GET | `/api/datasets/{dataset_id}/schema` | DuckDB Schema |
| GET | `/api/datasets/{dataset_id}/preview` | 最多 5 行预览 |
| POST | `/api/query` | 手动只读 SQL |
| POST | `/api/agent/query` | 单步骤 Text-to-SQL |
| POST | `/api/agent/analyze` | 多步骤分析、报告和图表 |
| GET | `/api/agent/analyze/{thread_id}/state` | 最新状态摘要 |
| GET | `/api/agent/analyze/{thread_id}/history` | Checkpoint 历史摘要 |

## 完整演示流程

项目提供 `sample_data/sales_demo.csv`：

```csv
region,product,sales,profit,sale_date
华东,A,120,20,2026-01-01
华南,B,90,-5,2026-01-02
华东,C,80,10,2026-01-03
华北,D,60,8,2026-01-04
```

以下命令中的 `dataset_xxx` 和 `analysis_xxx` 需要替换为上一步响应中的真实值。

### 1. 上传数据

```bash
curl -X POST http://127.0.0.1:8000/api/datasets/upload \
  -F "file=@sample_data/sales_demo.csv"
```

### 2. 查看 Profile、Schema 和 Preview

```bash
curl http://127.0.0.1:8000/api/datasets/dataset_xxx
curl http://127.0.0.1:8000/api/datasets/dataset_xxx/schema
curl http://127.0.0.1:8000/api/datasets/dataset_xxx/preview
```

### 3. 手动 SQL

```bash
curl -X POST http://127.0.0.1:8000/api/query \
  -H "Content-Type: application/json" \
  -d '{
    "dataset_id": "dataset_xxx",
    "sql": "SELECT region, SUM(sales) AS total_sales FROM dataset_xxx GROUP BY region ORDER BY total_sales DESC"
  }'
```

### 4. 单步骤 Agent

```bash
curl -X POST http://127.0.0.1:8000/api/agent/query \
  -H "Content-Type: application/json" \
  -d '{
    "dataset_id": "dataset_xxx",
    "question": "各地区总销售额分别是多少？"
  }'
```

### 5. 多步骤分析和图表

```bash
curl -X POST http://127.0.0.1:8000/api/agent/analyze \
  -H "Content-Type: application/json" \
  -d '{
    "dataset_id": "dataset_xxx",
    "question": "分析各地区销售表现，并找出利润最低的商品。"
  }'
```

响应包含：

- `thread_id`；
- 规划摘要和每个步骤结果；
- 综合报告；
- `chart` 或 `chart_error`；
- 成功、失败步骤数量和整体状态。

PNG 文件路径位于 `chart.file_path`。当前没有图表下载 API。

### 6. State 和 History

```bash
curl http://127.0.0.1:8000/api/agent/analyze/analysis_xxx/state
curl http://127.0.0.1:8000/api/agent/analyze/analysis_xxx/history
```

### 7. 重启后查询旧 thread

停止并重新启动 Uvicorn，保持 `CHECKPOINT_DB_PATH` 不变，再次调用上面的 State 和
History 接口即可读取旧 checkpoint。旧 `dataset_id` 不会恢复，因为 DatasetService
和 DuckDB 注册仍保存在内存中。

## 测试与验收

```bash
python -m compileall app tests
pytest -q
python -m pip check
```

当前验收结果：

```text
134 passed
No broken requirements found.
```

`tests/test_project_acceptance.py` 使用确定性的 Mock LLM 和模拟销售数据完整验证：上传、
Profile、手动 SQL、单步骤 Agent、多步骤分析、PNG、State、History 和应用重启后读取旧
thread。测试不会发送真实网络请求，也不会写入默认 checkpoint 数据库。

## 当前限制

- 数据集元数据和 DuckDB 注册只在当前进程内，服务重启后不会恢复；
- Excel 当前只分析第一个工作表和其中一个主要表格；低置信度结构需要人工核对 Preview；
- Checkpoint 持久化不等于 Resume，目前不能从中断节点继续；
- SQL 安全采用保守的词法规则，不是完整 SQL AST 沙箱；
- SQLite 适合单机原型，不适合高并发、多实例和高可用部署；
- 没有用户认证、权限隔离、配额、上传大小限制和审计日志；
- 本地页面面向单用户开发演示，不包含登录、会话共享或生产级文件访问控制；
- 没有结构化日志、请求关联 ID、指标或分布式追踪；
- Planner 步骤互相独立，不能消费前一步查询结果；
- 每次分析最多生成一张确定性 bar/line 图；
- 没有图表下载接口、历史清理或 checkpoint 删除；
- 没有 Resume、Interrupt、人工审批、异步 Worker、Redis 或 PostgreSQL；
- DeepSeek API 的延迟、费用、余额和模型可用性取决于外部服务。

更详细的验收结论见 `ACCEPTANCE_REPORT.md`，分阶段开发记录见 `WORKLOG.md`。