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

技术架构

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 的状态快照。

项目结构

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 子图

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 父图

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，默认数据库为：

.datapilot/checkpoints/checkpoints.sqlite

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

MCP 工具

MCP Streamable HTTP 服务挂载在：

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。

安装与配置

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

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

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 终端中，应分两行执行：

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

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。

启动服务：

uvicorn app.main:app --reload

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

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

本地操作页面

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

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：

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. 上传数据

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

2. 查看 Profile、Schema 和 Preview

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

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

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

5. 多步骤分析和图表

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

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 注册仍保存在内存中。

测试与验收

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

当前验收结果：

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。